clover's git

the-knife

1	import { AgentScope } from "@t3tools/shared/AgentScope";
2	import * as KeyedLock from "@t3tools/shared/KeyedLock";
3	import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
4	import {
5	ModelSelection,
6	OrchestrationV2DomainEvent,
7	OrchestrationV2ProviderSession,
8	type OrchestrationV2ProviderThread,
9	OrchestrationV2RuntimeRequest,
10	ProviderInstanceId,
11	ProviderSessionId,
12	ThreadId,
13	type ProviderThreadId,
14	} from "@t3tools/contracts";
15	import * as Cause from "effect/Cause";
16	import * as Clock from "effect/Clock";
17	import * as Context from "effect/Context";
18	import * as DateTime from "effect/DateTime";
19	import * as Duration from "effect/Duration";
20	import * as Effect from "effect/Effect";
21	import * as Exit from "effect/Exit";
22	import * as Fiber from "effect/Fiber";
23	import * as FiberSet from "effect/FiberSet";
24	import * as FileSystem from "effect/FileSystem";
25	import * as Layer from "effect/Layer";
26	import * as Option from "effect/Option";
27	import * as Queue from "effect/Queue";
28	import * as Ref from "effect/Ref";
29	import * as Schedule from "effect/Schedule";
30	import * as Schema from "effect/Schema";
31	import * as Scope from "effect/Scope";
32	import * as Semaphore from "effect/Semaphore";
33	import * as Stream from "effect/Stream";
34
35	import { normalizeModelMetricLabel } from "../observability/Attributes.ts";
36	import {
37	providerSessionsTotal,
38	providerTurnDuration,
39	providerTurnsTotal,
40	withMetrics,
41	} from "../observability/Metrics.ts";
42	import { ProviderWorkspaceMissingError } from "../provider/Errors.ts";
43	import * as ProjectService from "../project/ProjectService.ts";
44	import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
45	import * as ServerSettings from "../serverSettings.ts";
46	import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
47	import * as EventSink from "./EventSink.ts";
48	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
49	import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
50	import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
51	import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
52	import * as ProjectionStore from "./ProjectionStore.ts";
53
54	const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000;
55	const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000;
56	const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000;
57
58	const busyTurnPrefix = (providerThreadId: ProviderThreadId) => `${providerThreadId}#`;
59	/** The identity a turn's start and its `turn.terminal` share. */
60	const busyTurnKey = (providerThreadId: ProviderThreadId, runOrdinal: number) =>
61	`${busyTurnPrefix(providerThreadId)}${runOrdinal}`;
62	const UNLOAD_THREAD_TIMEOUT_MS = 10 * 1000;
63
64	export const ProviderSessionReleaseReason = Schema.Literals([
65	"idle_timeout",
66	"runtime_error",
67	"manual_shutdown",
68	"server_shutdown",
69	]);
70	export type ProviderSessionReleaseReason = typeof ProviderSessionReleaseReason.Type;
71
72	/**
73	* ProviderSessionManager owns live session residency: open sessions, idle release,
74	* explicit shutdown, and release-on-runtime-failure.
75	*
76	* It intentionally does not resurrect persisted sessions. Process-loss recovery
77	* terminalizes provider-bound work and retires non-replayable effects; a later
78	* user command or durable replay-safe operation opens a session lazily.
79	*/
80	export class ProviderSessionOpenError extends Schema.TaggedError<ProviderSessionOpenError>()(
81	"ProviderSessionOpenError",
82	{
83	instanceId: ProviderInstanceId,
84	providerSessionId: ProviderSessionId,
85	cause: Schema.optional(Schema.Defect()),
86	},
87	) {
88	override get message(): string {
89	return `Failed to open provider instance ${this.instanceId} session ${this.providerSessionId}.`;
90	}
91	}
92
93	export class ProviderSessionLookupError extends Schema.TaggedError<ProviderSessionLookupError>()(
94	"ProviderSessionLookupError",
95	{
96	providerSessionId: ProviderSessionId,
97	cause: Schema.optional(Schema.Defect()),
98	},
99	) {
100	override get message(): string {
101	return `Failed to look up provider session ${this.providerSessionId}.`;
102	}
103	}
104
105	export class ProviderSessionCloseError extends Schema.TaggedError<ProviderSessionCloseError>()(
106	"ProviderSessionCloseError",
107	{
108	providerSessionId: ProviderSessionId,
109	cause: Schema.optional(Schema.Defect()),
110	},
111	) {
112	override get message(): string {
113	return `Failed to close provider session ${this.providerSessionId}.`;
114	}
115	}
116
117	export class ProviderSessionReleaseError extends Schema.TaggedError<ProviderSessionReleaseError>()(
118	"ProviderSessionReleaseError",
119	{
120	providerSessionId: ProviderSessionId,
121	reason: ProviderSessionReleaseReason,
122	cause: Schema.optional(Schema.Defect()),
123	},
124	) {
125	override get message(): string {
126	return `Failed to release provider session ${this.providerSessionId}.`;
127	}
128	}
129
130	export class ProviderSessionActivityError extends Schema.TaggedError<ProviderSessionActivityError>()(
131	"ProviderSessionActivityError",
132	{
133	providerSessionId: ProviderSessionId,
134	cause: Schema.optional(Schema.Defect()),
135	},
136	) {
137	override get message(): string {
138	return `Failed to update provider session activity for ${this.providerSessionId}.`;
139	}
140	}
141
142	export const ProviderSessionManagerV2Error = Schema.Union([
143	ProviderSessionOpenError,
144	ProviderWorkspaceMissingError,
145	ProviderSessionLookupError,
146	ProviderSessionCloseError,
147	ProviderSessionReleaseError,
148	ProviderSessionActivityError,
149	]);
150	export type ProviderSessionManagerV2Error = typeof ProviderSessionManagerV2Error.Type;
151
152	export interface ProviderSessionManagerV2Shape {
153	readonly shutdown: Effect.Effect<void>;
154	readonly open: (input: {
155	readonly threadId: ThreadId;
156	readonly providerSessionId: ProviderSessionId;
157	readonly modelSelection: ModelSelection;
158	readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy;
159	readonly resumeFromSession?: OrchestrationV2ProviderSession;
160	readonly initialNativeThreadId?: string;
161	readonly initialProviderItemIdentityVersion?: 2;
162	}) => Effect.Effect<
163	ProviderAdapter.ProviderAdapterV2SessionRuntime,
164	ProviderSessionManagerV2Error
165	>;
166	readonly get: (
167	providerSessionId: ProviderSessionId,
168	) => Effect.Effect<
169	Option.Option<ProviderAdapter.ProviderAdapterV2SessionRuntime>,
170	ProviderSessionManagerV2Error
171	>;
172	/** Whether the session's runtime is running. Unlike `get`, not activity. */
173	readonly isLive: (providerSessionId: ProviderSessionId) => Effect.Effect<boolean>;
174	readonly close: (
175	providerSessionId: ProviderSessionId,
176	) => Effect.Effect<void, ProviderSessionManagerV2Error>;
177	/** Closes every live runtime owned by one provider instance. */
178	readonly closeInstance: (
179	instanceId: ProviderInstanceId,
180	) => Effect.Effect<void, ProviderSessionManagerV2Error>;
181	readonly release: (input: {
182	readonly providerSessionId: ProviderSessionId;
183	readonly reason: ProviderSessionReleaseReason;
184	readonly detail?: string;
185	}) => Effect.Effect<void, ProviderSessionManagerV2Error>;
186	readonly detach: (input: {
187	readonly providerSessionId: ProviderSessionId;
188	readonly threadId: ThreadId;
189	readonly detail?: string;
190	/**
191	* True for terminal detaches (thread archived or deleted): the thread's
192	* MCP credentials are revoked immediately instead of surviving for a
193	* potential re-attach.
194	*/
195	readonly revokeMcpCredential?: boolean;
196	}) => Effect.Effect<void, ProviderSessionManagerV2Error>;
197	}
198
199	export class ProviderSessionManagerV2 extends Context.Service<
200	ProviderSessionManagerV2,
201	ProviderSessionManagerV2Shape
202	>()("t3/orchestration-v2/ProviderSessionManager/ProviderSessionManagerV2") {}
203
204	interface LiveSessionEntry {
205	readonly attachedThreadIds: ReadonlySet<ThreadId>;
206	readonly loadedProviderThreadKeyByThread: ReadonlyMap<ThreadId, string>;
207	/**
208	* MCP credential session id issued for each attached thread. Revocation on
209	* detach/release is scoped to these ids so tearing down a superseded
210	* session cannot revoke a replacement session's credential for the same
211	* thread (the workspace-handoff sequence opens the replacement before the
212	* outbox executes the old session's detach).
213	*/
214	readonly mcpCredentialIdByThread: ReadonlyMap<ThreadId, string>;
215	readonly supportsMultipleProviderThreads: boolean;
216	readonly runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime;
217	readonly exposedRuntime: ProviderAdapter.ProviderAdapterV2SessionRuntime;
218	readonly eventSubscribers: Ref.Ref<
219	ReadonlyMap<number, Queue.Queue<ProviderSessionEventSignal, Cause.Done>>
220	>;
221	readonly requestEventPermit: Semaphore.Semaphore;
222	readonly scope: Scope.Closeable;
223	readonly idleGeneration: number;
224	/**
225	* Turns this session is running, keyed by `busyTurnKey`. A turn's start adds
226	* it and its `turn.terminal` (or a failed start) removes it, so a turn can
227	* only clear itself and the session is idle when the set is empty.
228	*/
229	readonly busyTurns: ReadonlySet<string>;
230	readonly lastActivityAtMs: number;
231	readonly idleFiber: Fiber.Fiber<void, never> | null;
232	/** Set when idle release is deferred for pending background work; bounds total deferral. */
233	readonly pinnedSinceMs: number | null;
234	/**
235	* Shared runtimes only: the provider thread each attached app thread last
236	* started a turn on, and the timer that unloads it once it has been idle
237	* for `idleTimeoutMs`.
238	*/
239	readonly idleThreadUnloads: ReadonlyMap<ThreadId, IdleThreadUnload>;
240	}
241
242	interface IdleThreadUnload {
243	readonly providerThread: OrchestrationV2ProviderThread;
244	readonly generation: number;
245	readonly fiber: Fiber.Fiber<void, never> | null;
246	}
247
248	type ProviderSessionEventSignal =
249	| { readonly type: "event"; readonly event: ProviderAdapter.ProviderAdapterV2Event }
250	| {
251	readonly type: "failure";
252	readonly cause: Cause.Cause<ProviderAdapter.ProviderAdapterV2Error>;
253	};
254
255	export interface ProviderSessionManagerV2LayerOptions {
256	readonly idleTimeoutMs?: number;
257	/** Cap on how long idle release may be deferred for pending background work. */
258	readonly maxIdlePinMs?: number;
259	/** Test replay harnesses can omit T3's MCP server from provider protocol fixtures. */
260	readonly configureMcp?: boolean;
261	}
262
263	function releaseStatusFor(
264	reason: ProviderSessionReleaseReason,
265	): OrchestrationV2ProviderSession["status"] {
266	return reason === "runtime_error" ? "error" : "stopped";
267	}
268
269	function releasedRuntimeRequestStatusFor(
270	reason: ProviderSessionReleaseReason,
271	): OrchestrationV2RuntimeRequest["status"] {
272	return reason === "manual_shutdown" || reason === "server_shutdown" ? "cancelled" : "expired";
273	}
274
275	function sessionKey(providerSessionId: ProviderSessionId): string {
276	return String(providerSessionId);
277	}
278
279	/**
280	* Runtime requests with no provider turn belong to the live session itself.
281	* Their node and transcript item are runless too, so they bypass the normal
282	* per-run subscriber and are persisted by the session event pump.
283	*/
284	function sessionScopedRuntimeRequestThreadId(
285	event: ProviderAdapter.ProviderAdapterV2Event,
286	): ThreadId | undefined {
287	switch (event.type) {
288	case "runtime_request.updated":
289	return event.runtimeRequest.providerTurnId === null ? event.threadId : undefined;
290	case "node.updated":
291	return event.node.runId === null && event.node.runtimeRequestId !== null
292	? event.node.threadId
293	: undefined;
294	case "turn_item.updated":
295	return event.turnItem.runId === null &&
296	(event.turnItem.type === "approval_request" || event.turnItem.type === "user_input_request")
297	? event.turnItem.threadId
298	: undefined;
299	default:
300	return undefined;
301	}
302	}
303
304	function providerThreadRuntimeKey(
305	providerThread: Parameters<
306	ProviderAdapter.ProviderAdapterV2SessionRuntime["resumeThread"]
307	>[0]["providerThread"],
308	): string {
309	const nativeThreadRef = providerThread.nativeThreadRef;
310	return nativeThreadRef === null
311	? String(providerThread.id)
312	: `${nativeThreadRef.driver}:${nativeThreadRef.nativeId}`;
313	}
314
315	function providerThreadLoadKey(input: {
316	readonly providerThread: Parameters<
317	ProviderAdapter.ProviderAdapterV2SessionRuntime["resumeThread"]
318	>[0]["providerThread"];
319	readonly modelSelection?: ModelSelection;
320	readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy;
321	}): string {
322	return JSON.stringify({
323	providerThread: providerThreadRuntimeKey(input.providerThread),
324	modelSelection: input.modelSelection ?? null,
325	runtimePolicy: input.runtimePolicy ?? null,
326	});
327	}
328
329	export const layerWithOptions = (
330	options: ProviderSessionManagerV2LayerOptions = {},
331	): Layer.Layer<
332	ProviderSessionManagerV2,
333	never,
334	| EventSink.EventSinkV2
335	| FileSystem.FileSystem
336	| IdAllocator.IdAllocatorV2
337	| McpProviderSessions.McpProviderSessions
338	| McpSessionRegistry.McpSessionRegistry
339	| ProjectionStore.ProjectionStoreV2
340	| ProviderEventIngestor.ProviderEventIngestorV2
341	| ProviderAdapterRegistry.ProviderAdapterRegistryV2
342	> =>
343	Layer.effect(
344	ProviderSessionManagerV2,
345	Effect.gen(function* () {
346	const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
347	const fileSystem = yield* FileSystem.FileSystem;
348	const mcpSessionRegistry = yield* McpSessionRegistry.McpSessionRegistry;
349	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
350	/**
351	* Optional so the many focused tests that assemble this layer by hand do
352	* not each need a settings stub; the production composition always
353	* provides it. When present, an unreadable settings file withholds
354	* browser access rather than granting it — an explicit "off" silently
355	* becoming "on" would violate the user's stated choice, whereas the
356	* reverse costs an agent one toolset and is visible immediately (#7083).
357	*/
358	const serverSettings = yield* Effect.serviceOption(ServerSettings.ServerSettingsService);
359	const projectService = yield* Effect.serviceOption(ProjectService.ProjectService);
360	const eventSink = yield* EventSink.EventSinkV2;
361	const idAllocator = yield* IdAllocator.IdAllocatorV2;
362	const providerEventIngestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
363	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
364	const agentScope = yield* AgentScope;
365	const agentAccessSettings = Effect.fn("ProviderSessionManagerV2.agentAccessSettings")(
366	function* (threadId: ThreadId) {
367	if (Option.isNone(serverSettings)) return { browser: true, device: false };
368	return yield* Effect.gen(function* () {
369	const settings = yield* serverSettings.value.getSettings;
370	const thread = yield* projectionStore.getThread(threadId);
371	const entries = Object.values(settings.projectSettingsOverrides);
372	const browserOverridden = entries.some(
373	(entry) => entry.enableAgentBrowserAccess !== undefined,
374	);
375	const deviceOverridden = entries.some(
376	(entry) => entry.enableAgentDeviceAccess !== undefined,
377	);
378	if (browserOverridden || deviceOverridden) {
379	const project = Option.isSome(projectService)
380	? yield* projectService.value.getById(thread.projectId)
381	: Option.none();
382	if (Option.isNone(project))
383	return {
384	browser: browserOverridden ? false : settings.enableAgentBrowserAccess,
385	device: deviceOverridden ? false : settings.enableAgentDeviceAccess,
386	};
387	}
388	const effective = resolveProjectSettings(settings, thread.projectId).settings;
389	return {
390	browser: effective.enableAgentBrowserAccess,
391	device: effective.enableAgentDeviceAccess,
392	};
393	}).pipe(
394	Effect.catch((cause) =>
395	Effect.logWarning(
396	"Could not resolve agent access; withholding browser and device tools.",
397	{ threadId, cause },
398	).pipe(Effect.as({ browser: false, device: false })),
399	),
400	);
401	},
402	);
403	const layerScope = yield* Effect.scope;
404	// Ctrl+C, or a stop that signals the whole process group, reaches the
405	// provider CLIs with the server. They report their own background work
406	// stopped before shutdown captures restart continuations, so provider
407	// events after the signal are dropped; restart recovery owns that state.
408	const shutdownSignal = { received: false };
409	const onShutdownSignal = () => {
410	shutdownSignal.received = true;
411	};
412	yield* Effect.acquireRelease(
413	Effect.sync(() => {
414	process.on("SIGINT", onShutdownSignal);
415	process.on("SIGTERM", onShutdownSignal);
416	}),
417	() =>
418	Effect.sync(() => {
419	process.off("SIGINT", onShutdownSignal);
420	process.off("SIGTERM", onShutdownSignal);
421	}),
422	);
423	const sessions = yield* Ref.make(new Map<string, LiveSessionEntry>());
424	// One retry per released entry, so a later release with the same id
425	// cannot drop cleanup for threads only the earlier session served.
426	const releaseRecordRetries = yield* FiberSet.make();
427	const nextSubscriberId = yield* Ref.make(0);
428	const sessionOpen = yield* KeyedLock.make<ProviderSessionId>();
429	// Orders a thread's attach against a detach unloading it on the same session.
430	const threadAttachment = yield* KeyedLock.make<string>();
431	const threadAttachmentKey = (input: {
432	readonly providerSessionId: ProviderSessionId;
433	readonly threadId: ThreadId;
434	}) => `${input.providerSessionId}\u0000${input.threadId}`;
435	const idleTimeoutMs = Math.max(1, options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS);
436	const maxIdlePinMs = Math.max(0, options.maxIdlePinMs ?? DEFAULT_MAX_IDLE_PIN_MS);
437	interface PreparedMcpCredential {
438	readonly mcpCredentialId: string | undefined;
439	/** True when this call minted the credential (vs reusing a live one). */
440	readonly issued: boolean;
441	}
442	/**
443	* Reservations protect a credential between prepareMcpSession handing it
444	* out and the owning session entry becoming visible in `sessions`.
445	* Adapters like ACP and OpenCode consume the credential eagerly during
446	* openSession, so a racing release must not revoke it in that window
447	* (rotating afterwards cannot repair an already-configured process).
448	* The holder MUST drop the reservation once the entry is recorded or the
449	* open fails.
450	*/
451	const mcpCredentialReservations = new Map<string, number>();
452	const mcpReservationKey = (threadId: ThreadId, mcpCredentialId: string) =>
453	`${threadId}\0${mcpCredentialId}`;
454	const reserveMcpCredential = (threadId: ThreadId, mcpCredentialId: string) => {
455	const key = mcpReservationKey(threadId, mcpCredentialId);
456	mcpCredentialReservations.set(key, (mcpCredentialReservations.get(key) ?? 0) + 1);
457	};
458	const dropMcpCredentialReservation = (threadId: ThreadId, mcpCredentialId: string) => {
459	const key = mcpReservationKey(threadId, mcpCredentialId);
460	const count = mcpCredentialReservations.get(key) ?? 0;
461	if (count <= 1) {
462	mcpCredentialReservations.delete(key);
463	} else {
464	mcpCredentialReservations.set(key, count - 1);
465	}
466	};
467	const isMcpCredentialReserved = (threadId: ThreadId, mcpCredentialId: string) =>
468	(mcpCredentialReservations.get(mcpReservationKey(threadId, mcpCredentialId)) ?? 0) > 0;
469	const mcpPrepareLock = yield* KeyedLock.make<ThreadId>();
470	/**
471	* Resolves (or mints) the thread's MCP credential and returns it with a
472	* reservation held; the caller must drop the reservation exactly once.
473	* Serialized per thread so two concurrent prepares cannot interleave
474	* their rotate steps and revoke each other's freshly minted credential.
475	*/
476	const prepareMcpSession = (
477	threadId: ThreadId,
478	providerInstanceId: ProviderInstanceId,
479	): Effect.Effect<PreparedMcpCredential> =>
480	options.configureMcp === false
481	? mcpSessions
482	.clear(threadId)
483	.pipe(Effect.as<PreparedMcpCredential>({ mcpCredentialId: undefined, issued: false }))
484	: mcpPrepareLock.withLock(
485	threadId,
486	Effect.gen(function* () {
487	// Reuse a still-valid credential for this thread instead of
488	// rotating: long-lived provider processes (codex app-server)
489	// build their MCP client once per conversation and keep using
490	// the credential it started with, so a thread that detaches and
491	// re-attaches across a workspace handoff must come back to the
492	// same token or the process's tool calls fail auth.
493	const { browser: browserToolsAvailable, device: deviceToolsAvailable } =
494	yield* agentAccessSettings(threadId);
495	const capabilities = new Set<
496	import("../mcp/McpInvocationContext.ts").McpCapability
497	>(["orchestration", "worktree", "pull-requests"]);
498	if (browserToolsAvailable) capabilities.add("preview");
499	if (deviceToolsAvailable) capabilities.add("device");
500	const existing = yield* mcpSessions.read(threadId);
501	if (existing !== undefined) {
502	// Reserve before the async resolve so a release cannot
503	// revoke the credential between validation and reservation.
504	reserveMcpCredential(threadId, existing.providerSessionId);
505	const rawToken = existing.authorizationHeader.replace(/^Bearer\s+/, "");
506	// The caller only learns of the reservation once this returns,
507	// so a stop while resolving must drop it here.
508	const resolved = yield* mcpSessionRegistry
509	.resolve(rawToken)
510	.pipe(
511	Effect.onInterrupt(() =>
512	Effect.sync(() =>
513	dropMcpCredentialReservation(threadId, existing.providerSessionId),
514	),
515	),
516	);
517	if (
518	resolved !== undefined &&
519	resolved.thread.threadId === threadId &&
520	resolved.thread.providerInstanceId === providerInstanceId &&
521	// A flipped browser-access setting must not survive through
522	// credential reuse: rotate so the new scope reflects it.
523	resolved.capabilities.has("preview") === browserToolsAvailable &&
524	resolved.capabilities.has("device") === deviceToolsAvailable
525	) {
526	return { mcpCredentialId: existing.providerSessionId, issued: false };
527	}
528	dropMcpCredentialReservation(threadId, existing.providerSessionId);
529	}
530	yield* mcpSessionRegistry.revokeThread(threadId);
531	const credential = yield* mcpSessionRegistry.issue({
532	threadId,
533	providerInstanceId,
534	browserToolsAvailable,
535	capabilities,
536	});
537	yield* mcpSessions.set(credential.config);
538	reserveMcpCredential(threadId, credential.config.providerSessionId);
539	return { mcpCredentialId: credential.config.providerSessionId, issued: true };
540	}),
541	);
542	/**
543	* With a credential id, revocation is scoped to that credential and the
544	* config slot is cleared only while it still holds it; a replacement
545	* session's newer credential survives. Without one (attach failed before
546	* a credential was recorded), fall back to thread-wide revocation.
547	*/
548	const clearMcpSession = (threadId: ThreadId, mcpCredentialId?: string) =>
549	mcpCredentialId === undefined
550	? mcpSessionRegistry
551	.revokeThread(threadId)
552	.pipe(Effect.tap(() => mcpSessions.clear(threadId)))
553	: mcpSessionRegistry.revokeProviderSession(mcpCredentialId).pipe(
554	Effect.tap(() =>
555	Effect.gen(function* () {
556	if ((yield* mcpSessions.read(threadId))?.providerSessionId === mcpCredentialId) {
557	yield* mcpSessions.clear(threadId);
558	}
559	}),
560	),
561	);
562
563	const publishToSubscribers = (
564	subscribers: Ref.Ref<
565	ReadonlyMap<number, Queue.Queue<ProviderSessionEventSignal, Cause.Done>>
566	>,
567	signal: ProviderSessionEventSignal,
568	) =>
569	Ref.get(subscribers).pipe(
570	Effect.flatMap((current) =>
571	Effect.forEach(current.values(), (queue) => Queue.offer(queue, signal), {
572	discard: true,
573	}),
574	),
575	);
576
577	const failSubscribers = (entry: LiveSessionEntry, detail: string) =>
578	Effect.gen(function* () {
579	const error = new ProviderAdapter.ProviderAdapterEventStreamError({
580	driver: entry.runtime.driver,
581	providerSessionId: entry.runtime.providerSessionId,
582	cause: detail,
583	});
584	const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map());
585	yield* Effect.forEach(
586	subscribers.values(),
587	(queue) =>
588	Queue.offer(queue, {
589	type: "failure",
590	cause: Cause.fail(error),
591	}),
592	{ discard: true },
593	);
594	});
595
596	const closeSubscribers = (entry: LiveSessionEntry) =>
597	Effect.gen(function* () {
598	const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map());
599	yield* Effect.forEach(
600	subscribers.values(),
601	(queue) => Queue.clear(queue).pipe(Effect.andThen(Queue.end(queue))),
602	{ discard: true },
603	);
604	});
605
606	// Preserve already-published terminal events while ending subscriptions.
607	// Server shutdown intentionally clears them; a provider-announced Stop
608	// must let consumers drain them before the stream completes.
609	const endSubscribers = (entry: LiveSessionEntry) =>
610	Effect.gen(function* () {
611	const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map());
612	yield* Effect.forEach(subscribers.values(), (queue) => Queue.end(queue), {
613	discard: true,
614	});
615	});
616
617	const cancelIdleFiber = (fiber: Fiber.Fiber<void, never> | null) =>
618	fiber === null ? Effect.void : Fiber.interrupt(fiber).pipe(Effect.ignore);
619
620	const writeProviderSessionEvents = (input: {
621	readonly runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime;
622	readonly threadIds: Iterable<ThreadId>;
623	readonly type: "provider-session.attached" | "provider-session.updated";
624	readonly payload: OrchestrationV2ProviderSession;
625	}) =>
626	Effect.gen(function* () {
627	const now = yield* DateTime.now;
628	const events = yield* Effect.forEach(input.threadIds, (threadId) =>
629	Effect.gen(function* () {
630	return {
631	id: yield* idAllocator.allocate.event({
632	threadId,
633	providerSessionId: input.runtime.providerSessionId,
634	}),
635	type: input.type,
636	threadId,
637	driver: input.runtime.driver,
638	providerInstanceId: input.runtime.instanceId,
639	occurredAt: now,
640	payload: input.payload,
641	} satisfies OrchestrationV2DomainEvent;
642	}),
643	);
644	if (events.length > 0) {
645	yield* eventSink.write({ events });
646	}
647	});
648
649	const writeReleasedSessionEvents = (input: {
650	readonly entry: LiveSessionEntry;
651	readonly reason: ProviderSessionReleaseReason;
652	readonly detail?: string;
653	}) =>
654	Effect.gen(function* () {
655	const now = yield* DateTime.now;
656	const payload: OrchestrationV2ProviderSession = {
657	...input.entry.runtime.providerSession,
658	status: releaseStatusFor(input.reason),
659	updatedAt: now,
660	lastError:
661	input.reason === "runtime_error"
662	? (input.detail ?? "Provider runtime failed.")
663	: null,
664	};
665	yield* writeProviderSessionEvents({
666	runtime: input.entry.runtime,
667	threadIds: input.entry.attachedThreadIds,
668	type: "provider-session.updated",
669	payload,
670	});
671	});
672
673	const writeReleasedRuntimeRequestEvents = (input: {
674	readonly entry: LiveSessionEntry;
675	readonly reason: ProviderSessionReleaseReason;
676	/** Requests created later belong to a replacement session with the same id. */
677	readonly releasedAt: DateTime.Utc;
678	}) =>
679	Effect.gen(function* () {
680	const providerSessionId = input.entry.runtime.providerSessionId;
681	const now = yield* DateTime.now;
682	const status = releasedRuntimeRequestStatusFor(input.reason);
683	const reason =
684	input.reason === "runtime_error"
685	? "Provider session failed before this runtime request was resolved."
686	: "Provider session was closed before this runtime request was resolved.";
687
688	const events: Array<OrchestrationV2DomainEvent> = [];
689	for (const threadId of input.entry.attachedThreadIds) {
690	const projection = yield* projectionStore.getThreadRecords(
691	threadId,
692	["runtimeRequests", "nodes", "turnItems"],
693	{ turnItemTypes: ["approval_request", "user_input_request"] },
694	);
695	const releasedRequests = projection.runtimeRequests.filter(
696	(request) =>
697	request.status === "pending" &&
698	request.responseCapability.type === "live" &&
699	request.responseCapability.providerSessionId === providerSessionId &&
700	DateTime.isLessThanOrEqualTo(request.createdAt, input.releasedAt),
701	);
702
703	for (const request of releasedRequests) {
704	events.push({
705	id: yield* idAllocator.allocate.event({
706	threadId,
707	providerSessionId,
708	}),
709	type: "runtime-request.updated",
710	threadId,
711	nodeId: request.nodeId,
712	driver: input.entry.runtime.driver,
713	occurredAt: now,
714	payload: {
715	...request,
716	status,
717	responseCapability: {
718	type: "not_resumable",
719	reason,
720	},
721	resolvedAt: now,
722	},
723	});
724
725	const requestNode = projection.nodes.find((node) => node.id === request.nodeId);
726	if (requestNode !== undefined) {
727	events.push({
728	id: yield* idAllocator.allocate.event({
729	threadId,
730	providerSessionId,
731	}),
732	type: "node.updated",
733	threadId,
734	...(requestNode.runId === null ? {} : { runId: requestNode.runId }),
735	nodeId: requestNode.id,
736	driver: input.entry.runtime.driver,
737	occurredAt: now,
738	payload: {
739	...requestNode,
740	status: input.reason === "runtime_error" ? "failed" : "cancelled",
741	completedAt: now,
742	},
743	});
744	}
745
746	const turnItem = projection.turnItems.find(
747	(item) =>
748	(item.type === "approval_request" || item.type === "user_input_request") &&
749	item.requestId === request.id,
750	);
751	if (turnItem !== undefined) {
752	events.push({
753	id: yield* idAllocator.allocate.event({
754	threadId,
755	providerSessionId,
756	}),
757	type: "turn-item.updated",
758	threadId,
759	...(turnItem.runId === null ? {} : { runId: turnItem.runId }),
760	...(turnItem.nodeId === null ? {} : { nodeId: turnItem.nodeId }),
761	driver: input.entry.runtime.driver,
762	occurredAt: now,
763	payload: {
764	...turnItem,
765	status: input.reason === "runtime_error" ? "failed" : "cancelled",
766	completedAt: now,
767	updatedAt: now,
768	},
769	});
770	}
771	}
772	}
773
774	if (events.length > 0) {
775	yield* eventSink.write({ events });
776	}
777	});
778
779	// Reads the background tasks a released session's process left on its
780	// provider threads. Read as soon as the session leaves the live map:
781	// after that, a replacement session with the same id can open and report
782	// tasks of its own on the same provider threads. Server shutdown keeps
783	// rosters, since startup recovery clears them and records the cancelled
784	// work. Only runtimes whose tasks end with their process opt in (Claude,
785	// Muse). OpenCode's background subagents run on a server shared by its
786	// sessions and can outlive this one, so their roster stays.
787	const readReleasedBackgroundTasks = (input: {
788	readonly entry: LiveSessionEntry;
789	readonly reason: ProviderSessionReleaseReason;
790	readonly detachedThreadId?: ThreadId;
791	}) =>
792	Effect.gen(function* () {
793	if (
794	input.reason === "server_shutdown" ||
795	input.entry.runtime.backgroundTasksEndWithProcess !== true
796	) {
797	return [];
798	}
799	const providerSessionId = input.entry.runtime.providerSessionId;
800	// A detach that released the session already removed its thread
801	// from the attached set.
802	const threadIds = new Set(input.entry.attachedThreadIds);
803	if (input.detachedThreadId !== undefined) threadIds.add(input.detachedThreadId);
804	const released = [];
805	for (const threadId of threadIds) {
806	const projection = yield* projectionStore.getThreadRecords(threadId, [
807	"providerThreads",
808	]);
809	for (const providerThread of projection.providerThreads) {
810	const tasks = providerThread.pendingBackgroundTasks ?? [];
811	if (providerThread.providerSessionId !== providerSessionId || tasks.length === 0) {
812	continue;
813	}
814	released.push({
815	threadId,
816	providerSessionId,
817	providerThreadId: providerThread.id,
818	taskIds: new Set(tasks.map((task) => task.taskId)),
819	});
820	}
821	}
822	return released;
823	}).pipe(
824	Effect.catchCause((cause) =>
825	Effect.logWarning("orchestration-v2.provider-session-release-roster-read-failed", {
826	providerSessionId: input.entry.runtime.providerSessionId,
827	cause,
828	}).pipe(Effect.as([])),
829	),
830	);
831	type ReleasedBackgroundTasks = Effect.Success<
832	ReturnType<typeof readReleasedBackgroundTasks>
833	>[number];
834
835	// Removes the tasks `readReleasedBackgroundTasks` found from their
836	// rosters. Their process is gone, so nothing will report them ending,
837	// and a leftover roster keeps the thread Waiting (#15239). Tasks a
838	// replacement session reported since are left in place. The write
839	// commits only if the stored row still equals what was read, and is
840	// retried a few times when a provider update lands in between.
841	const writeReleasedBackgroundWorkEvents = (
842	released: ReadonlyArray<ReleasedBackgroundTasks>,
843	) =>
844	Effect.forEach(
845	released,
846	({ threadId, providerSessionId, providerThreadId, taskIds }) =>
847	Effect.gen(function* () {
848	for (let attempt = 0; attempt < 3; attempt++) {
849	const projection = yield* projectionStore.getThreadRecords(threadId, [
850	"providerThreads",
851	]);
852	const providerThread = projection.providerThreads.find(
853	(candidate) => candidate.id === providerThreadId,
854	);
855	const tasks = providerThread?.pendingBackgroundTasks ?? [];
856	const remaining = tasks.filter((task) => !taskIds.has(task.taskId));
857	if (providerThread === undefined || remaining.length === tasks.length) return;
858	const now = yield* DateTime.now;
859	const { committed } = yield* eventSink.writeIfProviderThreadUnchanged({
860	expected: providerThread,
861	events: [
862	{
863	id: yield* idAllocator.allocate.event({ threadId, providerSessionId }),
864	type: "provider-thread.updated",
865	threadId,
866	driver: providerThread.driver,
867	providerInstanceId: providerThread.providerInstanceId,
868	occurredAt: now,
869	payload: {
870	...providerThread,
871	pendingBackgroundTasks: remaining,
872	updatedAt: now,
873	},
874	},
875	],
876	});
877	if (committed) return;
878	}
879	yield* Effect.logWarning(
880	"orchestration-v2.provider-session-release-roster-conflict",
881	{
882	providerSessionId,
883	providerThreadId,
884	},
885	);
886	}),
887	{ discard: true },
888	);
889
890	// Records a released session as stopped and resolves the live runtime
891	// requests and background-task rosters it left. Each write runs even if
892	// another fails. Once a replacement session opens with the same id, it
893	// owns the session status, so only the requests are settled.
894	const writeReleaseRecords = (input: {
895	readonly entry: LiveSessionEntry;
896	readonly reason: ProviderSessionReleaseReason;
897	readonly detail?: string;
898	readonly releasedBackgroundTasks: ReadonlyArray<ReleasedBackgroundTasks>;
899	readonly releasedAt: DateTime.Utc;
900	readonly replaced: boolean;
901	}) =>
902	Effect.all(
903	[
904	input.replaced
905	? Effect.succeed(Exit.void)
906	: Effect.exit(writeReleasedSessionEvents(input)),
907	Effect.exit(
908	writeReleasedRuntimeRequestEvents(input).pipe(
909	input.entry.requestEventPermit.withPermits(1),
910	),
911	),
912	Effect.exit(writeReleasedBackgroundWorkEvents(input.releasedBackgroundTasks)),
913	],
914	{ concurrency: 1 },
915	).pipe(Effect.flatMap(Exit.asVoidAll));
916
917	// The session already left the live map, so a later release finds
918	// nothing to do. Without a retry the UI would keep a ready session and
919	// answerable approvals until a server restart. Each attempt holds the
920	// session's open lock, so it sees a replacement that opened meanwhile.
921	const retryReleaseRecords = (
922	input: Omit<Parameters<typeof writeReleaseRecords>[0], "replaced">,
923	) => {
924	const providerSessionId = input.entry.runtime.providerSessionId;
925	const attempt = Effect.gen(function* () {
926	const exit = yield* Effect.exit(
927	sessionOpen.withLock(
928	providerSessionId,
929	Effect.gen(function* () {
930	const replaced = (yield* Ref.get(sessions)).has(sessionKey(providerSessionId));
931	yield* writeReleaseRecords({ ...input, replaced });
932	}),
933	),
934	);
935	if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) return yield* exit;
936	yield* Effect.logWarning("orchestration-v2.provider-session-release-records-failed", {
937	providerSessionId,
938	cause: exit.cause,
939	});
940	// A failed SQL commit is a defect, so every failure but interruption
941	// is retried.
942	return yield* Effect.fail(exit.cause);
943	});
944	return attempt.pipe(
945	Effect.retry({
946	schedule: Schedule.exponential("1 second").pipe(
947	Schedule.modifyDelay(({ duration }) =>
948	Effect.succeed(Duration.min(duration, Duration.seconds(30))),
949	),
950	),
951	}),
952	Effect.delay("1 second"),
953	FiberSet.run(releaseRecordRetries),
954	);
955	};
956
957	const logReleaseFailure =
958	(providerSessionId: ProviderSessionId) =>
959	<E, R>(release: Effect.Effect<void, E, R>) =>
960	release.pipe(
961	Effect.catchCause((cause) =>
962	Effect.logWarning("orchestration-v2.provider-session-release-failed", {
963	providerSessionId,
964	cause,
965	}),
966	),
967	);
968
969	// Removes the live entry and reads the request cleanup cutoff while
970	// holding the entry's request permit. A request the event pump is
971	// persisting for this runtime lands before the cutoff, and once the
972	// entry is gone the pump persists no more for it. A replacement's
973	// requests come after its own open.
974	const removeLiveEntry = (input: {
975	readonly providerSessionId: ProviderSessionId;
976	readonly onlyIfIdleGeneration?: number;
977	}): Effect.Effect<readonly [Option.Option<LiveSessionEntry>, DateTime.Utc]> =>
978	Effect.gen(function* () {
979	const key = sessionKey(input.providerSessionId);
980	const candidate = (yield* Ref.get(sessions)).get(key);
981	if (candidate === undefined) {
982	return [Option.none<LiveSessionEntry>(), yield* DateTime.now] as const;
983	}
984	const removed = yield* Effect.zip(
985	Ref.modify(sessions, (current) => {
986	const existing = current.get(key);
987	if (existing !== candidate) {
988	return [existing === undefined ? "gone" : "changed", current] as const;
989	}
990	if (
991	input.onlyIfIdleGeneration !== undefined &&
992	(existing.busyTurns.size > 0 ||
993	existing.idleGeneration !== input.onlyIfIdleGeneration)
994	) {
995	return ["kept", current] as const;
996	}
997	const updated = new Map(current);
998	updated.delete(key);
999	return ["removed", updated] as const;
1000	}),
1001	DateTime.now,
1002	).pipe(candidate.requestEventPermit.withPermits(1));
1003	const [outcome, releasedAt] = removed;
1004	// Another entry took this id while the permit was held; release it instead.
1005	if (outcome === "changed") return yield* removeLiveEntry(input);
1006	return [
1007	outcome === "removed" ? Option.some(candidate) : Option.none<LiveSessionEntry>(),
1008	releasedAt,
1009	] as const;
1010	});
1011
1012	// Scope close can wedge on a misbehaving adapter finalizer (e.g. a
1013	// provider process that never yields its message stream). Time-box it so
1014	// the caller, and any lock or worker it holds, moves on and leaves a
1015	// diagnosable trail. A close that finishes late is still logged.
1016	const closeScopeWithin = (
1017	scope: Scope.Closeable,
1018	annotations: { readonly providerSessionId?: ProviderSessionId; readonly reason: string },
1019	) =>
1020	Effect.gen(function* () {
1021	const closeFiber = yield* Scope.close(scope, Exit.void).pipe(
1022	Effect.exit,
1023	Effect.forkDetach({ startImmediately: true }),
1024	);
1025	const closeExit = yield* Fiber.join(closeFiber).pipe(
1026	Effect.timeoutOption(RELEASE_SCOPE_CLOSE_TIMEOUT_MS),
1027	);
1028	if (Option.isNone(closeExit)) {
1029	yield* Effect.logWarning("orchestration-v2.provider-session-scope-close-timeout", {
1030	...annotations,
1031	timeoutMs: RELEASE_SCOPE_CLOSE_TIMEOUT_MS,
1032	});
1033	yield* Fiber.join(closeFiber).pipe(
1034	Effect.flatMap((exit) =>
1035	Exit.isFailure(exit)
1036	? Effect.logWarning("orchestration-v2.provider-session-scope-close-failed", {
1037	...annotations,
1038	cause: exit.cause,
1039	})
1040	: Effect.logInfo(
1041	"orchestration-v2.provider-session-scope-close-completed-late",
1042	annotations,
1043	),
1044	),
1045	Effect.forkDetach,
1046	);
1047	}
1048	return closeExit;
1049	});
1050
1051	const releaseEntry = (input: {
1052	readonly providerSessionId: ProviderSessionId;
1053	readonly reason: ProviderSessionReleaseReason;
1054	readonly detail?: string;
1055	/** The thread whose detach released the session; it is no longer attached. */
1056	readonly detachedThreadId?: ThreadId;
1057	readonly cancelIdleFiber?: boolean;
1058	readonly onlyIfIdleGeneration?: number;
1059	readonly gracefulSubscribers?: boolean;
1060	}) =>
1061	Effect.acquireUseRelease(
1062	removeLiveEntry(input),
1063	([entry, releasedAt]) =>
1064	Option.match(entry, {
1065	onNone: () => Effect.void,
1066	onSome: (entry) =>
1067	Effect.gen(function* () {
1068	const releasedBackgroundTasks = yield* readReleasedBackgroundTasks({
1069	entry,
1070	reason: input.reason,
1071	...(input.detachedThreadId === undefined
1072	? {}
1073	: { detachedThreadId: input.detachedThreadId }),
1074	});
1075	if (input.cancelIdleFiber !== false) {
1076	yield* cancelIdleFiber(entry.idleFiber);
1077	}
1078	if (input.gracefulSubscribers === true) {
1079	yield* endSubscribers(entry);
1080	} else if (input.reason === "server_shutdown") {
1081	yield* closeSubscribers(entry);
1082	} else {
1083	yield* failSubscribers(
1084	entry,
1085	input.detail ?? `Provider session released: ${input.reason}.`,
1086	);
1087	}
1088	const closeExit = yield* closeScopeWithin(entry.scope, {
1089	providerSessionId: input.providerSessionId,
1090	reason: input.reason,
1091	});
1092	const records = {
1093	entry,
1094	reason: input.reason,
1095	...(input.detail === undefined ? {} : { detail: input.detail }),
1096	releasedBackgroundTasks,
1097	releasedAt,
1098	};
1099	const recorded = yield* Effect.exit(
1100	writeReleaseRecords({ ...records, replaced: false }),
1101	);
1102	if (Exit.isFailure(recorded)) {
1103	yield* retryReleaseRecords(records);
1104	return yield* recorded;
1105	}
1106	if (Option.isSome(closeExit) && Exit.isFailure(closeExit.value)) {
1107	return yield* Effect.failCause(closeExit.value.cause);
1108	}
1109	}).pipe(
1110	withMetrics({
1111	counter: providerSessionsTotal,
1112	attributes: {
1113	provider: entry.runtime.driver,
1114	operation: "release",
1115	reason: input.reason,
1116	},
1117	}),
1118	),
1119	}),
1120	([entry]) =>
1121	Option.match(entry, {
1122	onNone: () => Effect.void,
1123	onSome: (entry) =>
1124	// Revoke every credential this session recorded, including for
1125	// threads that detached without re-attaching: the provider
1126	// process is gone, so nothing holds them anymore. Skip threads
1127	// a live replacement session took over, since credential reuse
1128	// means the replacement may hold this very credential.
1129	Ref.get(sessions).pipe(
1130	Effect.flatMap((current) =>
1131	Effect.forEach(
1132	entry.mcpCredentialIdByThread,
1133	([threadId, mcpCredentialId]) => {
1134	// Id-sensitive: a stale record for the same thread but
1135	// a DIFFERENT credential (left behind by an old session
1136	// the thread rotated away from) must not veto revoking
1137	// this session's own credential, or it leaks forever.
1138	// A reservation means an in-flight open is configuring
1139	// a provider process with this credential right now;
1140	// revoking it here would strand that process (eager
1141	// adapters cannot pick up a rotated token).
1142	const heldElsewhere =
1143	isMcpCredentialReserved(threadId, mcpCredentialId) ||
1144	Array.from(current.values()).some(
1145	(other) =>
1146	other !== entry &&
1147	(other.attachedThreadIds.has(threadId) ||
1148	other.mcpCredentialIdByThread.get(threadId) === mcpCredentialId),
1149	);
1150	return heldElsewhere
1151	? Effect.void
1152	: clearMcpSession(threadId, mcpCredentialId);
1153	},
1154	{ discard: true },
1155	),
1156	),
1157	),
1158	}),
1159	).pipe(
1160	Effect.catchCause((cause) =>
1161	Effect.fail(
1162	new ProviderSessionReleaseError({
1163	providerSessionId: input.providerSessionId,
1164	reason: input.reason,
1165	cause,
1166	}),
1167	),
1168	),
1169	);
1170
1171	// Annotated to break the releaseIfStillIdle <-> scheduleIdleReleaseInternal
1172	// inference cycle introduced by the pin re-arm below.
1173	const releaseIfStillIdle = (input: {
1174	readonly providerSessionId: ProviderSessionId;
1175	readonly generation: number;
1176	}): Effect.Effect<void> =>
1177	Effect.gen(function* () {
1178	const current = yield* Ref.get(sessions);
1179	const key = sessionKey(input.providerSessionId);
1180	const entry = current.get(key);
1181	if (
1182	entry === undefined ||
1183	entry.busyTurns.size > 0 ||
1184	entry.idleGeneration !== input.generation
1185	) {
1186	return;
1187	}
1188	// Capture runtime identity before yielding: a replacement session
1189	// can reuse the same providerSessionId while this fiber is parked.
1190	const probedRuntime = entry.runtime;
1191	const hasPendingWork =
1192	probedRuntime.hasPendingBackgroundWork === undefined
1193	? false
1194	: yield* probedRuntime.hasPendingBackgroundWork.pipe(
1195	Effect.catchCause(() => Effect.succeed(false)),
1196	);
1197	if (hasPendingWork) {
1198	const now = yield* Clock.currentTimeMillis;
1199	const pinnedSinceMs = entry.pinnedSinceMs ?? now;
1200	if (now - pinnedSinceMs < maxIdlePinMs) {
1201	const shouldContinuePin = yield* Ref.modify(sessions, (latest) => {
1202	const latestEntry = latest.get(key);
1203	if (
1204	latestEntry === undefined ||
1205	latestEntry.busyTurns.size > 0 ||
1206	latestEntry.idleGeneration !== input.generation ||
1207	latestEntry.runtime !== probedRuntime
1208	) {
1209	return [false, latest] as const;
1210	}
1211	const updated = new Map(latest);
1212	updated.set(key, { ...latestEntry, pinnedSinceMs });
1213	return [true, updated] as const;
1214	});
1215	if (!shouldContinuePin) {
1216	// Generation or runtime advanced while we probed pending work;
1217	// the current owner of the entry owns idle release.
1218	return;
1219	}
1220	yield* Effect.logInfo("orchestration-v2.driver-session.idle-release-deferred", {
1221	providerSessionId: input.providerSessionId,
1222	pinnedForMs: now - pinnedSinceMs,
1223	});
1224	// Re-check on this fiber after another idle window. Do not call
1225	// scheduleIdleReleaseInternal: that cancels entry.idleFiber, which
1226	// is this fiber, and can self-deadlock on Fiber.interrupt.
1227	yield* Effect.sleep(Duration.millis(idleTimeoutMs));
1228	return yield* releaseIfStillIdle(input);
1229	}
1230	yield* Effect.logWarning("orchestration-v2.driver-session.idle-release-pin-expired", {
1231	providerSessionId: input.providerSessionId,
1232	pinnedForMs: now - pinnedSinceMs,
1233	});
1234	}
1235	// hasPendingBackgroundWork yields to the adapter, so the idle
1236	// decision above can go stale; the generation guard revalidates
1237	// busyTurns and idleGeneration inside releaseEntry's atomic
1238	// entry removal.
1239	yield* releaseEntry({
1240	providerSessionId: input.providerSessionId,
1241	reason: "idle_timeout",
1242	cancelIdleFiber: false,
1243	onlyIfIdleGeneration: input.generation,
1244	}).pipe(
1245	Effect.catchCause((cause) =>
1246	Effect.logWarning("orchestration-v2.driver-session.idle-release-failed", {
1247	providerSessionId: input.providerSessionId,
1248	cause,
1249	}),
1250	),
1251	);
1252	});
1253
1254	const withActivityError = <A, E, R>(
1255	providerSessionId: ProviderSessionId,
1256	effect: Effect.Effect<A, E, R>,
1257	): Effect.Effect<A, ProviderSessionActivityError, R> =>
1258	effect.pipe(
1259	Effect.catchCause((cause) =>
1260	Effect.fail(
1261	new ProviderSessionActivityError({
1262	providerSessionId,
1263	cause,
1264	}),
1265	),
1266	),
1267	);
1268
1269	// Provider activity restarts the pin budget, so background work that keeps
1270	// reporting progress (a long Claude workflow) is never cut off; only work
1271	// that went silent for maxIdlePinMs is released.
1272	const scheduleIdleReleaseInternal = (
1273	providerSessionId: ProviderSessionId,
1274	options?: { readonly providerActivity?: boolean },
1275	) =>
1276	Effect.gen(function* () {
1277	const key = sessionKey(providerSessionId);
1278	const current = yield* Ref.get(sessions);
1279	const entry = current.get(key);
1280	if (entry === undefined || entry.busyTurns.size > 0) {
1281	return;
1282	}
1283
1284	yield* cancelIdleFiber(entry.idleFiber);
1285	const generation = entry.idleGeneration + 1;
1286	const idleFiber = yield* Effect.sleep(Duration.millis(idleTimeoutMs)).pipe(
1287	Effect.andThen(releaseIfStillIdle({ providerSessionId, generation })),
1288	Effect.forkIn(layerScope),
1289	);
1290	const lastActivityAtMs = yield* Clock.currentTimeMillis;
1291	yield* Ref.update(sessions, (latest) => {
1292	const latestEntry = latest.get(key);
1293	if (latestEntry === undefined || latestEntry.busyTurns.size > 0) {
1294	return latest;
1295	}
1296	const updated = new Map(latest);
1297	updated.set(key, {
1298	...latestEntry,
1299	idleGeneration: generation,
1300	idleFiber,
1301	lastActivityAtMs,
1302	...(options?.providerActivity === true ? { pinnedSinceMs: null } : {}),
1303	});
1304	return updated;
1305	});
1306	});
1307
1308	const scheduleIdleRelease = (providerSessionId: ProviderSessionId) =>
1309	withActivityError(providerSessionId, scheduleIdleReleaseInternal(providerSessionId));
1310
1311	const touchActivity = (
1312	providerSessionId: ProviderSessionId,
1313	options?: { readonly providerActivity?: boolean },
1314	) =>
1315	withActivityError(
1316	providerSessionId,
1317	Effect.gen(function* () {
1318	const lastActivityAtMs = yield* Clock.currentTimeMillis;
1319	yield* Ref.update(sessions, (current) => {
1320	const entry = current.get(sessionKey(providerSessionId));
1321	if (entry === undefined) {
1322	return current;
1323	}
1324	const updated = new Map(current);
1325	updated.set(sessionKey(providerSessionId), {
1326	...entry,
1327	lastActivityAtMs,
1328	});
1329	return updated;
1330	});
1331	yield* scheduleIdleReleaseInternal(providerSessionId, options);
1332	}),
1333	);
1334
1335	/** Returns the runtime the thread was attached to, or undefined if it already was. */
1336	const attachThread = (input: {
1337	readonly providerSessionId: ProviderSessionId;
1338	readonly threadId: ThreadId;
1339	}) =>
1340	withActivityError(
1341	input.providerSessionId,
1342	Ref.modify(sessions, (current) => {
1343	const entry = current.get(sessionKey(input.providerSessionId));
1344	if (entry === undefined || entry.attachedThreadIds.has(input.threadId)) {
1345	return [undefined, current] as const;
1346	}
1347	const updated = new Map(current);
1348	updated.set(sessionKey(input.providerSessionId), {
1349	...entry,
1350	attachedThreadIds: new Set([...entry.attachedThreadIds, input.threadId]),
1351	});
1352	return [entry.runtime, updated] as const;
1353	}),
1354	);
1355
1356	/**
1357	* Undoes an attach to `runtime`. A replacement session that reopened under
1358	* the same id since is left alone.
1359	*/
1360	const removeThreadAttachment = (input: {
1361	readonly providerSessionId: ProviderSessionId;
1362	readonly threadId: ThreadId;
1363	readonly runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime;
1364	}) =>
1365	Ref.update(sessions, (current) => {
1366	const key = sessionKey(input.providerSessionId);
1367	const entry = current.get(key);
1368	if (
1369	entry === undefined ||
1370	entry.runtime !== input.runtime ||
1371	!entry.attachedThreadIds.has(input.threadId)
1372	) {
1373	return current;
1374	}
1375	const attachedThreadIds = new Set(entry.attachedThreadIds);
1376	attachedThreadIds.delete(input.threadId);
1377	const loadedProviderThreadKeyByThread = new Map(entry.loadedProviderThreadKeyByThread);
1378	loadedProviderThreadKeyByThread.delete(input.threadId);
1379	const updated = new Map(current);
1380	updated.set(key, {
1381	...entry,
1382	attachedThreadIds,
1383	loadedProviderThreadKeyByThread,
1384	});
1385	return updated;
1386	});
1387
1388	const isProviderThreadLoaded = (input: {
1389	readonly providerSessionId: ProviderSessionId;
1390	readonly threadId: ThreadId;
1391	readonly providerThreadKey: string;
1392	}) =>
1393	Ref.get(sessions).pipe(
1394	Effect.map(
1395	(current) =>
1396	current
1397	.get(sessionKey(input.providerSessionId))
1398	?.loadedProviderThreadKeyByThread.get(input.threadId) === input.providerThreadKey,
1399	),
1400	);
1401
1402	const markProviderThreadLoaded = (input: {
1403	readonly providerSessionId: ProviderSessionId;
1404	readonly threadId: ThreadId;
1405	readonly providerThreadKey: string;
1406	}) =>
1407	Ref.update(sessions, (current) => {
1408	const key = sessionKey(input.providerSessionId);
1409	const entry = current.get(key);
1410	if (entry === undefined) {
1411	return current;
1412	}
1413	const loadedProviderThreadKeyByThread = new Map(entry.loadedProviderThreadKeyByThread);
1414	loadedProviderThreadKeyByThread.set(input.threadId, input.providerThreadKey);
1415	const updated = new Map(current);
1416	updated.set(key, { ...entry, loadedProviderThreadKeyByThread });
1417	return updated;
1418	});
1419
1420	const ensureThreadAttached = (input: {
1421	readonly providerSessionId: ProviderSessionId;
1422	readonly threadId: ThreadId;
1423	readonly providerInstanceId: ProviderInstanceId;
1424	}) =>
1425	Effect.suspend(() => {
1426	let attachedTo: ProviderAdapter.ProviderAdapterV2SessionRuntime | undefined;
1427	let preparedForCleanup: PreparedMcpCredential | undefined;
1428	let reservationDropped = false;
1429	const dropReservation = () => {
1430	if (!reservationDropped && preparedForCleanup?.mcpCredentialId !== undefined) {
1431	reservationDropped = true;
1432	dropMcpCredentialReservation(input.threadId, preparedForCleanup.mcpCredentialId);
1433	}
1434	};
1435	// The whole attach, including undoing a failed one, holds the
1436	// thread's lock: a concurrent attach of the same thread waits, so it
1437	// never sees an attachment that this call is about to roll back.
1438	const attach = Effect.gen(function* () {
1439	const attached = yield* attachThread(input).pipe(
1440	// Recorded with no gap for an interrupt: cleanup undoes only an
1441	// attach this call made, never one an earlier open made.
1442	Effect.tap((runtime) => Effect.sync(() => (attachedTo = runtime))),
1443	Effect.uninterruptible,
1444	);
1445	if (attached !== undefined) {
1446	const prepared = yield* prepareMcpSession(input.threadId, input.providerInstanceId);
1447	preparedForCleanup = prepared;
1448	if (prepared.mcpCredentialId !== undefined) {
1449	const mcpCredentialId = prepared.mcpCredentialId;
1450	yield* Ref.update(sessions, (current) => {
1451	const key = sessionKey(input.providerSessionId);
1452	const entry = current.get(key);
1453	if (entry === undefined) return current;
1454	const mcpCredentialIdByThread = new Map(entry.mcpCredentialIdByThread);
1455	mcpCredentialIdByThread.set(input.threadId, mcpCredentialId);
1456	const updated = new Map(current);
1457	updated.set(key, { ...entry, mcpCredentialIdByThread });
1458	return updated;
1459	});
1460	}
1461	const entry = (yield* Ref.get(sessions)).get(sessionKey(input.providerSessionId));
1462	if (entry !== undefined) {
1463	yield* withActivityError(
1464	input.providerSessionId,
1465	writeProviderSessionEvents({
1466	runtime: entry.runtime,
1467	threadIds: [input.threadId],
1468	type: "provider-session.attached",
1469	payload: entry.runtime.providerSession,
1470	}),
1471	);
1472	}
1473	}
1474	}).pipe(
1475	// An interrupted attach is undone too, so the next attach writes
1476	// the attachment instead of finding the thread already attached.
1477	Effect.onError(() =>
1478	attachedTo === undefined
1479	? Effect.void
1480	: removeThreadAttachment({ ...input, runtime: attachedTo }).pipe(
1481	Effect.andThen(
1482	Effect.suspend(() => {
1483	dropReservation();
1484	// Revoke only a credential this attach freshly minted: a
1485	// REUSED credential is by definition held by another
1486	// live provider process, and revoking it thread-wide
1487	// would break that process's MCP client mid-conversation.
1488	if (preparedForCleanup?.issued !== true) return Effect.void;
1489	const mcpCredentialId = preparedForCleanup.mcpCredentialId;
1490	const attachedRuntime = attachedTo;
1491	// As in release: a replacement session (or an open
1492	// configuring one) may have taken the credential up.
1493	return Ref.get(sessions).pipe(
1494	Effect.flatMap((current) =>
1495	(mcpCredentialId !== undefined &&
1496	isMcpCredentialReserved(input.threadId, mcpCredentialId)) ||
1497	Array.from(current.values()).some(
1498	(other) =>
1499	other.runtime !== attachedRuntime &&
1500	(other.attachedThreadIds.has(input.threadId) ||
1501	(mcpCredentialId !== undefined &&
1502	other.mcpCredentialIdByThread.get(input.threadId) ===
1503	mcpCredentialId)),
1504	)
1505	? Effect.void
1506	: clearMcpSession(input.threadId, mcpCredentialId),
1507	),
1508	);
1509	}),
1510	),
1511	),
1512	),
1513	);
1514	return threadAttachment.withLock(threadAttachmentKey(input), attach).pipe(
1515	// The entry's own record (written above while the thread is
1516	// attached) guards the credential from here on; the reservation
1517	// is only needed until then. Ensuring covers defects/interrupts.
1518	Effect.ensuring(Effect.sync(dropReservation)),
1519	);
1520	});
1521
1522	const markBusy = (providerSessionId: ProviderSessionId, turnKey: string) =>
1523	withActivityError(
1524	providerSessionId,
1525	Effect.gen(function* () {
1526	const key = sessionKey(providerSessionId);
1527	const now = yield* Clock.currentTimeMillis;
1528	const idleFiber = yield* Ref.modify(sessions, (current) => {
1529	const entry = current.get(key);
1530	if (entry === undefined) {
1531	return [null, current] as const;
1532	}
1533	const updated = new Map(current);
1534	updated.set(key, {
1535	...entry,
1536	busyTurns: new Set(entry.busyTurns).add(turnKey),
1537	idleFiber: null,
1538	lastActivityAtMs: now,
1539	pinnedSinceMs: null,
1540	});
1541	return [entry.idleFiber, updated] as const;
1542	});
1543	yield* cancelIdleFiber(idleFiber);
1544	}),
1545	);
1546
1547	// Clearing a turn that is not marked busy (one whose failed start already
1548	// cleared it, or a subagent turn the manager never started) only
1549	// records activity.
1550	const markIdle = (
1551	providerSessionId: ProviderSessionId,
1552	providerThreadId: ProviderThreadId,
1553	runOrdinal: number,
1554	) =>
1555	withActivityError(
1556	providerSessionId,
1557	Effect.gen(function* () {
1558	const key = sessionKey(providerSessionId);
1559	const now = yield* Clock.currentTimeMillis;
1560	yield* Ref.update(sessions, (current) => {
1561	const entry = current.get(key);
1562	if (entry === undefined) {
1563	return current;
1564	}
1565	const busyTurns = new Set(entry.busyTurns);
1566	busyTurns.delete(busyTurnKey(providerThreadId, runOrdinal));
1567	const updated = new Map(current);
1568	updated.set(key, {
1569	...entry,
1570	busyTurns,
1571	lastActivityAtMs: now,
1572	});
1573	return updated;
1574	});
1575	yield* scheduleIdleReleaseInternal(providerSessionId);
1576	yield* scheduleThreadUnload(providerSessionId, providerThreadId);
1577	}),
1578	);
1579
1580	const hasBusyTurn = (entry: LiveSessionEntry, providerThreadId: ProviderThreadId) => {
1581	const prefix = busyTurnPrefix(providerThreadId);
1582	for (const turnKey of entry.busyTurns) {
1583	if (turnKey.startsWith(prefix)) return true;
1584	}
1585	return false;
1586	};
1587
1588	const updateIdleThreadUnload = (
1589	providerSessionId: ProviderSessionId,
1590	threadId: ThreadId,
1591	update: (current: IdleThreadUnload | undefined) => IdleThreadUnload | undefined,
1592	) =>
1593	Ref.modify(sessions, (current) => {
1594	const key = sessionKey(providerSessionId);
1595	const entry = current.get(key);
1596	if (entry === undefined) return [undefined, current] as const;
1597	const previous = entry.idleThreadUnloads.get(threadId);
1598	const next = update(previous);
1599	const idleThreadUnloads = new Map(entry.idleThreadUnloads);
1600	if (next === undefined) idleThreadUnloads.delete(threadId);
1601	else idleThreadUnloads.set(threadId, next);
1602	const updated = new Map(current);
1603	updated.set(key, { ...entry, idleThreadUnloads });
1604	return [previous, updated] as const;
1605	});
1606
1607	/**
1608	* Starts tracking the provider thread an app thread runs its turns on, and
1609	* stops any unload pending for it. Called before the thread is resumed or
1610	* given a turn, so an unload cannot land between a resume that found the
1611	* thread loaded and the turn that relies on it.
1612	*/
1613	const holdThreadLoaded = (input: {
1614	readonly providerSessionId: ProviderSessionId;
1615	readonly threadId: ThreadId;
1616	readonly providerThread: OrchestrationV2ProviderThread;
1617	}) =>
1618	Effect.gen(function* () {
1619	const entry = (yield* Ref.get(sessions)).get(sessionKey(input.providerSessionId));
1620	if (
1621	entry === undefined ||
1622	!entry.supportsMultipleProviderThreads ||
1623	entry.exposedRuntime.unloadThread === undefined ||
1624	input.providerThread.nativeThreadRef === null
1625	) {
1626	return;
1627	}
1628	const previous = yield* updateIdleThreadUnload(
1629	input.providerSessionId,
1630	input.threadId,
1631	(current) => ({
1632	providerThread: input.providerThread,
1633	generation: (current?.generation ?? 0) + 1,
1634	fiber: null,
1635	}),
1636	);
1637	yield* cancelIdleFiber(previous?.fiber ?? null);
1638	});
1639
1640	/**
1641	* A shared runtime never goes idle while any of its threads is in use, so
1642	* its threads get the idle timeout one by one: a thread with no turn for
1643	* `idleTimeoutMs` is unloaded from the runtime, along with the native MCP
1644	* servers it started. Its next turn's resume loads it again.
1645	*/
1646	const scheduleThreadUnload = (
1647	providerSessionId: ProviderSessionId,
1648	providerThreadId: ProviderThreadId,
1649	) =>
1650	Effect.gen(function* () {
1651	const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId));
1652	if (entry === undefined || hasBusyTurn(entry, providerThreadId)) return;
1653	const tracked = Array.from(entry.idleThreadUnloads).find(
1654	([, pending]) => pending.providerThread.id === providerThreadId,
1655	);
1656	if (tracked === undefined) return;
1657	const [threadId, pending] = tracked;
1658	const generation = pending.generation + 1;
1659	const fiber = yield* Effect.sleep(Duration.millis(idleTimeoutMs)).pipe(
1660	Effect.andThen(unloadIdleThread({ providerSessionId, threadId, generation })),
1661	Effect.forkIn(layerScope),
1662	);
1663	// A turn that started meanwhile already moved the generation on.
1664	const previous = yield* updateIdleThreadUnload(providerSessionId, threadId, (current) =>
1665	current?.generation === pending.generation
1666	? { ...current, generation, fiber }
1667	: current,
1668	);
1669	yield* cancelIdleFiber(
1670	previous?.generation === pending.generation ? previous.fiber : fiber,
1671	);
1672	});
1673
1674	const unloadIdleThread = (input: {
1675	readonly providerSessionId: ProviderSessionId;
1676	readonly threadId: ThreadId;
1677	readonly generation: number;
1678	}): Effect.Effect<void> =>
1679	Effect.gen(function* () {
1680	const outcome = yield* threadAttachment.withLock(
1681	threadAttachmentKey(input),
1682	Effect.gen(function* () {
1683	const key = sessionKey(input.providerSessionId);
1684	const entry = (yield* Ref.get(sessions)).get(key);
1685	const pending = entry?.idleThreadUnloads.get(input.threadId);
1686	const unloadThread = entry?.exposedRuntime.unloadThread;
1687	if (
1688	entry === undefined ||
1689	pending === undefined ||
1690	pending.generation !== input.generation ||
1691	unloadThread === undefined ||
1692	!entry.attachedThreadIds.has(input.threadId) ||
1693	hasBusyTurn(entry, pending.providerThread.id)
1694	) {
1695	return "skipped" as const;
1696	}
1697	// Unloading stops the native thread's background terminals, so a
1698	// thread still running background work stays loaded.
1699	const hasPendingWork =
1700	entry.runtime.hasPendingBackgroundWorkForThread === undefined
1701	? false
1702	: yield* entry.runtime
1703	.hasPendingBackgroundWorkForThread(pending.providerThread)
1704	.pipe(Effect.catchCause(() => Effect.succeed(false)));
1705	if (hasPendingWork) return "deferred" as const;
1706	const unloading = yield* Ref.modify(sessions, (current) => {
1707	const latest = current.get(key);
1708	if (
1709	latest?.runtime !== entry.runtime ||
1710	latest.idleThreadUnloads.get(input.threadId)?.generation !== input.generation
1711	) {
1712	return [false, current] as const;
1713	}
1714	const loadedProviderThreadKeyByThread = new Map(
1715	latest.loadedProviderThreadKeyByThread,
1716	);
1717	loadedProviderThreadKeyByThread.delete(input.threadId);
1718	const idleThreadUnloads = new Map(latest.idleThreadUnloads);
1719	idleThreadUnloads.delete(input.threadId);
1720	const updated = new Map(current);
1721	updated.set(key, {
1722	...latest,
1723	loadedProviderThreadKeyByThread,
1724	idleThreadUnloads,
1725	});
1726	return [true, updated] as const;
1727	});
1728	if (!unloading) return "skipped" as const;
1729	yield* unloadThread({ providerThread: pending.providerThread }).pipe(
1730	Effect.timeout(UNLOAD_THREAD_TIMEOUT_MS),
1731	Effect.catchCause((cause) =>
1732	Effect.logWarning("orchestration-v2.driver-session.idle-unload-failed", {
1733	providerSessionId: input.providerSessionId,
1734	threadId: input.threadId,
1735	providerThreadId: pending.providerThread.id,
1736	cause,
1737	}),
1738	),
1739	);
1740	return "unloaded" as const;
1741	}),
1742	);
1743	// Re-check on this fiber after another idle window, outside the
1744	// lock so the thread's next attach is not held up meanwhile.
1745	if (outcome === "deferred") {
1746	yield* Effect.sleep(Duration.millis(idleTimeoutMs));
1747	return yield* unloadIdleThread(input);
1748	}
1749	});
1750
1751	const observeActivity = (
1752	providerSessionId: ProviderSessionId,
1753	activity: Effect.Effect<void, ProviderSessionActivityError>,
1754	) =>
1755	activity.pipe(
1756	Effect.catchCause((cause) =>
1757	Effect.logWarning("orchestration-v2.driver-session.activity-failed", {
1758	providerSessionId,
1759	cause,
1760	}),
1761	),
1762	);
1763
1764	const makeEventSubscription = (
1765	subscribers: Ref.Ref<
1766	ReadonlyMap<number, Queue.Queue<ProviderSessionEventSignal, Cause.Done>>
1767	>,
1768	): Effect.Effect<ProviderAdapter.ProviderAdapterV2EventSubscription> =>
1769	Effect.gen(function* () {
1770	const queue = yield* Queue.unbounded<ProviderSessionEventSignal, Cause.Done>();
1771	const subscriberId = yield* Ref.getAndUpdate(nextSubscriberId, (value) => value + 1);
1772	yield* Ref.update(subscribers, (current) => {
1773	const updated = new Map(current);
1774	updated.set(subscriberId, queue);
1775	return updated;
1776	});
1777	const close = Ref.modify(subscribers, (current) => {
1778	if (!current.has(subscriberId)) {
1779	return [false, current] as const;
1780	}
1781	const updated = new Map(current);
1782	updated.delete(subscriberId);
1783	return [true, updated] as const;
1784	}).pipe(
1785	Effect.flatMap((removed) =>
1786	removed
1787	? Queue.clear(queue).pipe(Effect.andThen(Queue.end(queue)), Effect.asVoid)
1788	: Effect.void,
1789	),
1790	);
1791	const events = Stream.fromQueue(queue).pipe(
1792	Stream.mapEffect((signal) =>
1793	signal.type === "event"
1794	? Effect.succeed(signal.event)
1795	: Effect.failCause(signal.cause),
1796	),
1797	Stream.ensuring(close),
1798	);
1799	return { events, close } satisfies ProviderAdapter.ProviderAdapterV2EventSubscription;
1800	});
1801
1802	const decorateRuntime = (
1803	runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime,
1804	eventSubscribers: Ref.Ref<
1805	ReadonlyMap<number, Queue.Queue<ProviderSessionEventSignal, Cause.Done>>
1806	>,
1807	): ProviderAdapter.ProviderAdapterV2SessionRuntime => {
1808	const providerSessionId = runtime.providerSessionId;
1809	const subscribeEvents = makeEventSubscription(eventSubscribers);
1810	// Every provider's turn operations pass through here, so this is where they are
1811	// counted. Only turn starts are timed: until the provider accepts the turn.
1812	const turnMetrics = (operation: string, model?: string) =>
1813	withMetrics({
1814	counter: providerTurnsTotal,
1815	...(operation === "send" ? { timer: providerTurnDuration } : {}),
1816	attributes: {
1817	provider: runtime.driver,
1818	operation,
1819	modelFamily: normalizeModelMetricLabel(model),
1820	},
1821	});
1822	return {
1823	...runtime,
1824	subscribeEvents,
1825	events: Stream.unwrap(
1826	subscribeEvents.pipe(Effect.map((subscription) => subscription.events)),
1827	),
1828	ensureThread: (input) =>
1829	observeActivity(
1830	providerSessionId,
1831	ensureThreadAttached({
1832	providerSessionId,
1833	threadId: input.threadId,
1834	providerInstanceId: runtime.instanceId,
1835	}),
1836	).pipe(
1837	Effect.andThen(runtime.ensureThread(input)),
1838	Effect.tap((providerThread) =>
1839	markProviderThreadLoaded({
1840	providerSessionId,
1841	threadId: input.threadId,
1842	providerThreadKey: providerThreadLoadKey({
1843	providerThread,
1844	modelSelection: input.modelSelection,
1845	runtimePolicy: input.runtimePolicy,
1846	}),
1847	}),
1848	),
1849	),
1850	resumeThread: (input) => {
1851	const threadId = input.threadId ?? input.providerThread.appThreadId;
1852	if (threadId === null || threadId === undefined) {
1853	return runtime.resumeThread(input);
1854	}
1855	const providerThreadKey = providerThreadLoadKey({
1856	providerThread: input.providerThread,
1857	...(input.modelSelection === undefined
1858	? {}
1859	: { modelSelection: input.modelSelection }),
1860	...(input.runtimePolicy === undefined ? {} : { runtimePolicy: input.runtimePolicy }),
1861	});
1862	return observeActivity(
1863	providerSessionId,
1864	ensureThreadAttached({
1865	providerSessionId,
1866	threadId,
1867	providerInstanceId: runtime.instanceId,
1868	}),
1869	).pipe(
1870	Effect.andThen(
1871	holdThreadLoaded({
1872	providerSessionId,
1873	threadId,
1874	providerThread: input.providerThread,
1875	}),
1876	),
1877	Effect.andThen(
1878	isProviderThreadLoaded({ providerSessionId, threadId, providerThreadKey }),
1879	),
1880	Effect.flatMap((loaded) =>
1881	loaded ? Effect.succeed(input.providerThread) : runtime.resumeThread(input),
1882	),
1883	Effect.tap((providerThread) =>
1884	markProviderThreadLoaded({
1885	providerSessionId,
1886	threadId,
1887	providerThreadKey: providerThreadLoadKey({
1888	providerThread,
1889	...(input.modelSelection === undefined
1890	? {}
1891	: { modelSelection: input.modelSelection }),
1892	...(input.runtimePolicy === undefined
1893	? {}
1894	: { runtimePolicy: input.runtimePolicy }),
1895	}),
1896	}),
1897	),
1898	);
1899	},
1900	forkThread: (input) =>
1901	observeActivity(
1902	providerSessionId,
1903	ensureThreadAttached({
1904	providerSessionId,
1905	threadId: input.targetThreadId,
1906	providerInstanceId: runtime.instanceId,
1907	}),
1908	).pipe(
1909	Effect.andThen(runtime.forkThread(input)),
1910	Effect.tap((providerThread) =>
1911	markProviderThreadLoaded({
1912	providerSessionId,
1913	threadId: input.targetThreadId,
1914	providerThreadKey: providerThreadLoadKey({
1915	providerThread,
1916	...(input.modelSelection === undefined
1917	? {}
1918	: { modelSelection: input.modelSelection }),
1919	...(input.runtimePolicy === undefined
1920	? {}
1921	: { runtimePolicy: input.runtimePolicy }),
1922	}),
1923	}),
1924	),
1925	),
1926	startTurn: (input) =>
1927	observeActivity(
1928	providerSessionId,
1929	ensureThreadAttached({
1930	providerSessionId,
1931	threadId: input.threadId,
1932	providerInstanceId: runtime.instanceId,
1933	}),
1934	).pipe(
1935	// A start that fails or is stopped may never emit turn.terminal,
1936	// so it clears its own turn or the session never goes idle. If
1937	// the adapter emits the terminal anyway, clearing the same turn
1938	// again changes nothing, so another thread's turn on a shared
1939	// session stays busy either way.
1940	Effect.andThen(
1941	holdThreadLoaded({
1942	providerSessionId,
1943	threadId: input.threadId,
1944	providerThread: input.providerThread,
1945	}),
1946	),
1947	Effect.andThen(
1948	Effect.acquireUseRelease(
1949	observeActivity(
1950	providerSessionId,
1951	markBusy(
1952	providerSessionId,
1953	busyTurnKey(input.providerThread.id, input.runOrdinal),
1954	),
1955	),
1956	() =>
1957	runtime.startTurn(input).pipe(turnMetrics("send", input.modelSelection.model)),
1958	(_, exit) =>
1959	Exit.isFailure(exit)
1960	? observeActivity(
1961	providerSessionId,
1962	markIdle(providerSessionId, input.providerThread.id, input.runOrdinal),
1963	)
1964	: Effect.void,
1965	),
1966	),
1967	),
1968	steerTurn: (input) =>
1969	observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(
1970	Effect.andThen(runtime.steerTurn(input).pipe(turnMetrics("steer"))),
1971	),
1972	interruptTurn: (input) =>
1973	observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(
1974	Effect.andThen(runtime.interruptTurn(input).pipe(turnMetrics("interrupt"))),
1975	),
1976	respondToRuntimeRequest: (input) =>
1977	observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(
1978	Effect.andThen(
1979	runtime
1980	.respondToRuntimeRequest(input)
1981	.pipe(turnMetrics("runtime-request-response")),
1982	),
1983	),
1984	};
1985	};
1986
1987	const persistProviderSessionUpdate = (
1988	entry: LiveSessionEntry,
1989	event: Extract<
1990	ProviderAdapter.ProviderAdapterV2Event,
1991	{ readonly type: "provider_session.updated" }
1992	>,
1993	) =>
1994	Effect.gen(function* () {
1995	const current = (yield* Ref.get(sessions)).get(
1996	sessionKey(entry.runtime.providerSessionId),
1997	);
1998	if (current?.runtime !== entry.runtime) {
1999	return;
2000	}
2001	yield* writeProviderSessionEvents({
2002	runtime: entry.runtime,
2003	threadIds: current.attachedThreadIds,
2004	type: "provider-session.updated",
2005	payload: event.providerSession,
2006	});
2007	}).pipe(
2008	Effect.catchCause((cause) =>
2009	Effect.logWarning("orchestration-v2.driver-session.status-persist-failed", {
2010	providerSessionId: entry.runtime.providerSessionId,
2011	cause,
2012	}),
2013	),
2014	);
2015
2016	const startEventPump = (entry: LiveSessionEntry) => {
2017	let stoppedByProvider = false;
2018	return entry.runtime.events.pipe(
2019	Stream.runForEach((event) => {
2020	if (shutdownSignal.received) return Effect.void;
2021	if (
2022	event.type === "provider_session.updated" &&
2023	event.providerSession.status === "stopped"
2024	) {
2025	stoppedByProvider = true;
2026	}
2027	return observeActivity(
2028	entry.runtime.providerSessionId,
2029	event.type === "turn.terminal"
2030	? markIdle(
2031	entry.runtime.providerSessionId,
2032	event.providerThreadId,
2033	event.runOrdinal,
2034	)
2035	: touchActivity(entry.runtime.providerSessionId, { providerActivity: true }),
2036	).pipe(
2037	Effect.andThen(
2038	event.type === "provider_session.updated"
2039	? persistProviderSessionUpdate(entry, event)
2040	: Effect.void,
2041	),
2042	Effect.andThen(
2043	Effect.gen(function* () {
2044	// Some providers can block before a run subscriber exists
2045	// (project trust, login, or session-switch hooks). Persist
2046	// their runless request artifacts directly so the normal T3
2047	// request UI can answer them and unblock session setup.
2048	const threadId = sessionScopedRuntimeRequestThreadId(event);
2049	if (threadId !== undefined) {
2050	yield* Effect.gen(function* () {
2051	const current = (yield* Ref.get(sessions)).get(
2052	sessionKey(entry.runtime.providerSessionId),
2053	);
2054	if (current?.runtime !== entry.runtime) return;
2055	yield* providerEventIngestor
2056	.ingestNormalized({
2057	providerSessionId: entry.runtime.providerSessionId,
2058	providerInstanceId: entry.runtime.instanceId,
2059	threadId,
2060	event,
2061	})
2062	.pipe(
2063	Effect.mapError(
2064	(cause) =>
2065	new ProviderAdapter.ProviderAdapterEventStreamError({
2066	driver: entry.runtime.driver,
2067	providerSessionId: entry.runtime.providerSessionId,
2068	cause,
2069	}),
2070	),
2071	);
2072	}).pipe(entry.requestEventPermit.withPermits(1));
2073	return;
2074	}
2075	yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event });
2076	}),
2077	),
2078	);
2079	}),
2080	Effect.exit,
2081	Effect.flatMap((exit) =>
2082	Effect.gen(function* () {
2083	// A provider that exits on the shutdown signal is released by shutdown.
2084	if (shutdownSignal.received) return;
2085	const current = (yield* Ref.get(sessions)).get(
2086	sessionKey(entry.runtime.providerSessionId),
2087	);
2088	if (current?.runtime !== entry.runtime) {
2089	return;
2090	}
2091	if (stoppedByProvider && Exit.isSuccess(exit)) {
2092	yield* releaseEntry({
2093	providerSessionId: entry.runtime.providerSessionId,
2094	reason: "manual_shutdown",
2095	gracefulSubscribers: true,
2096	}).pipe(logReleaseFailure(entry.runtime.providerSessionId));
2097	return;
2098	}
2099	const cause = Exit.isFailure(exit)
2100	? exit.cause
2101	: Cause.fail(
2102	new ProviderAdapter.ProviderAdapterEventStreamError({
2103	driver: entry.runtime.driver,
2104	providerSessionId: entry.runtime.providerSessionId,
2105	cause: "Provider event stream ended unexpectedly.",
2106	}),
2107	);
2108	yield* publishToSubscribers(entry.eventSubscribers, {
2109	type: "failure",
2110	cause,
2111	});
2112	yield* Ref.set(entry.eventSubscribers, new Map());
2113	yield* releaseEntry({
2114	providerSessionId: entry.runtime.providerSessionId,
2115	reason: "runtime_error",
2116	detail: Cause.pretty(cause),
2117	}).pipe(logReleaseFailure(entry.runtime.providerSessionId));
2118	}),
2119	),
2120	Effect.forkIn(layerScope),
2121	);
2122	};
2123
2124	// Parent of every session scope. On layer close, shutdown releases the
2125	// live sessions first, then closes any session whose open is still in
2126	// flight, time-boxed so a stuck adapter cannot hold up server shutdown.
2127	// Both steps run in parallel. The service launcher kills the server
2128	// 5 s after it asks it to stop, and a busy server can hold 15 sessions.
2129	const sessionScopes = yield* Scope.make("parallel");
2130	const shutdown = Effect.gen(function* () {
2131	const activeSessions = [...(yield* Ref.get(sessions)).values()];
2132	yield* Effect.forEach(
2133	activeSessions,
2134	(entry) =>
2135	releaseEntry({
2136	providerSessionId: entry.runtime.providerSessionId,
2137	reason: "server_shutdown",
2138	}).pipe(
2139	Effect.catchCause((cause) =>
2140	Effect.logWarning("orchestration-v2.driver-session.shutdown-release-failed", {
2141	providerSessionId: entry.runtime.providerSessionId,
2142	cause,
2143	}),
2144	),
2145	),
2146	{ concurrency: "unbounded", discard: true },
2147	);
2148	});
2149	yield* Effect.addFinalizer(() =>
2150	shutdown.pipe(
2151	Effect.ensuring(closeScopeWithin(sessionScopes, { reason: "server_shutdown" })),
2152	),
2153	);
2154
2155	return ProviderSessionManagerV2.of({
2156	shutdown,
2157	open: (input) =>
2158	sessionOpen.withLock(
2159	input.providerSessionId,
2160	Effect.gen(function* () {
2161	const cwd = input.runtimePolicy.cwd;
2162	if (cwd !== null) {
2163	const workspaceIsDirectory = yield* fileSystem.stat(cwd).pipe(
2164	Effect.map((stat) => stat.type === "Directory"),
2165	Effect.catch((error) => Effect.succeed(error.reason._tag !== "NotFound")),
2166	);
2167	if (!workspaceIsDirectory) {
2168	return yield* new ProviderWorkspaceMissingError({
2169	threadId: input.threadId,
2170	cwd,
2171	});
2172	}
2173	}
2174	const key = sessionKey(input.providerSessionId);
2175	const existing = (yield* Ref.get(sessions)).get(key);
2176	if (existing !== undefined) {
2177	const attached = existing.attachedThreadIds.has(input.threadId);
2178	if (!attached && !existing.supportsMultipleProviderThreads) {
2179	return yield* new ProviderSessionOpenError({
2180	instanceId: input.modelSelection.instanceId,
2181	providerSessionId: input.providerSessionId,
2182	cause: `Provider ${existing.runtime.driver} does not support attaching multiple app threads to one session.`,
2183	});
2184	}
2185	// Joining a shared session is a new session for this thread.
2186	if (!attached) yield* agentScope.clear(input.threadId);
2187	yield* ensureThreadAttached({
2188	providerSessionId: input.providerSessionId,
2189	threadId: input.threadId,
2190	providerInstanceId: existing.runtime.instanceId,
2191	});
2192	yield* touchActivity(input.providerSessionId);
2193	return existing.exposedRuntime;
2194	}
2195
2196	const adapter = yield* registry.get(input.modelSelection.instanceId).pipe(
2197	Effect.mapError(
2198	(cause) =>
2199	new ProviderSessionOpenError({
2200	instanceId: input.modelSelection.instanceId,
2201	providerSessionId: input.providerSessionId,
2202	cause,
2203	}),
2204	),
2205	);
2206	const prepared = yield* prepareMcpSession(
2207	input.threadId,
2208	input.modelSelection.instanceId,
2209	);
2210	const mcpCredentialId = prepared.mcpCredentialId;
2211	// The reservation from prepare protects the credential (which
2212	// eager adapters bake into the provider process during
2213	// openSession) from racing releases until this session's entry
2214	// is recorded below. Dropped exactly once on every path.
2215	let reservationDropped = mcpCredentialId === undefined;
2216	const dropReservation = Effect.sync(() => {
2217	if (!reservationDropped && mcpCredentialId !== undefined) {
2218	reservationDropped = true;
2219	dropMcpCredentialReservation(input.threadId, mcpCredentialId);
2220	}
2221	});
2222	const sessionScope = yield* Scope.fork(sessionScopes);
2223	// The previous session's agent scope must not explain this
2224	// session's failures, even when this provider runs without one.
2225	yield* agentScope.clear(input.threadId);
2226	const runtime = yield* adapter
2227	.openSession({
2228	threadId: input.threadId,
2229	providerSessionId: input.providerSessionId,
2230	modelSelection: input.modelSelection,
2231	runtimePolicy: input.runtimePolicy,
2232	...(input.resumeFromSession === undefined
2233	? {}
2234	: { resumeFromSession: input.resumeFromSession }),
2235	...(input.initialNativeThreadId === undefined
2236	? {}
2237	: { initialNativeThreadId: input.initialNativeThreadId }),
2238	...(input.initialProviderItemIdentityVersion === undefined
2239	? {}
2240	: {
2241	initialProviderItemIdentityVersion:
2242	input.initialProviderItemIdentityVersion,
2243	}),
2244	})
2245	.pipe(
2246	Effect.provideService(Scope.Scope, sessionScope),
2247	// Any failure, including a Stop that interrupts a slow
2248	// handshake, stops the provider process this open started.
2249	// The session cleanup runs first, and the close is
2250	// time-boxed: this runs under the session's open lock, so an
2251	// adapter finalizer that never finishes must not hold the
2252	// interrupter, the lock, or later opens.
2253	Effect.onError(() =>
2254	dropReservation.pipe(
2255	// Clear only a session this open freshly set up: a reused
2256	// one is held by another live provider process and must
2257	// survive this open's failure.
2258	Effect.andThen(
2259	prepared.issued
2260	? clearMcpSession(input.threadId, mcpCredentialId)
2261	: Effect.void,
2262	),
2263	Effect.ensuring(
2264	closeScopeWithin(sessionScope, {
2265	providerSessionId: input.providerSessionId,
2266	reason: "open_failed",
2267	}),
2268	),
2269	),
2270	),
2271	Effect.mapError(
2272	(cause) =>
2273	new ProviderSessionOpenError({
2274	instanceId: input.modelSelection.instanceId,
2275	providerSessionId: input.providerSessionId,
2276	cause,
2277	}),
2278	),
2279	withMetrics({
2280	counter: providerSessionsTotal,
2281	attributes: { provider: adapter.driver, operation: "open" },
2282	}),
2283	);
2284	const eventSubscribers = yield* Ref.make<
2285	ReadonlyMap<number, Queue.Queue<ProviderSessionEventSignal, Cause.Done>>
2286	>(new Map());
2287	const exposedRuntime = decorateRuntime(runtime, eventSubscribers);
2288	const now = yield* Clock.currentTimeMillis;
2289	const entry: LiveSessionEntry = {
2290	attachedThreadIds: new Set([input.threadId]),
2291	loadedProviderThreadKeyByThread: new Map(),
2292	mcpCredentialIdByThread:
2293	mcpCredentialId === undefined
2294	? new Map()
2295	: new Map([[input.threadId, mcpCredentialId]]),
2296	supportsMultipleProviderThreads:
2297	runtime.providerSession.capabilities.sessions
2298	.supportsMultipleProviderThreadsPerSession,
2299	runtime,
2300	exposedRuntime,
2301	eventSubscribers,
2302	requestEventPermit: yield* Semaphore.make(1),
2303	scope: sessionScope,
2304	idleGeneration: 0,
2305	busyTurns: new Set(),
2306	lastActivityAtMs: now,
2307	idleFiber: null,
2308	pinnedSinceMs: null,
2309	idleThreadUnloads: new Map(),
2310	};
2311	yield* Ref.update(sessions, (current) => {
2312	const updated = new Map(current);
2313	updated.set(key, entry);
2314	return updated;
2315	});
2316	// The entry now guards the credential via its recorded id, so
2317	// the pre-open reservation can be dropped.
2318	yield* dropReservation;
2319	yield* withActivityError(
2320	input.providerSessionId,
2321	writeProviderSessionEvents({
2322	runtime,
2323	threadIds: [input.threadId],
2324	type: "provider-session.attached",
2325	payload: runtime.providerSession,
2326	}),
2327	).pipe(
2328	// Released on interrupt too: this entry has no event pump or
2329	// idle timer yet, so nothing else would ever release it.
2330	Effect.onError((cause) =>
2331	releaseEntry(
2332	Cause.hasInterruptsOnly(cause)
2333	? {
2334	providerSessionId: input.providerSessionId,
2335	reason: "manual_shutdown",
2336	detail: "The provider session start was interrupted.",
2337	}
2338	: {
2339	providerSessionId: input.providerSessionId,
2340	reason: "runtime_error",
2341	detail: "Failed to persist the provider-session attachment.",
2342	},
2343	).pipe(logReleaseFailure(input.providerSessionId)),
2344	),
2345	);
2346	yield* startEventPump(entry);
2347	yield* scheduleIdleRelease(input.providerSessionId);
2348	return exposedRuntime;
2349	}),
2350	),
2351	isLive: (providerSessionId) =>
2352	Ref.get(sessions).pipe(
2353	Effect.map((current) => current.has(sessionKey(providerSessionId))),
2354	),
2355	get: (providerSessionId) =>
2356	Effect.gen(function* () {
2357	const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId));
2358	if (entry === undefined) {
2359	return Option.none<ProviderAdapter.ProviderAdapterV2SessionRuntime>();
2360	}
2361	yield* touchActivity(providerSessionId);
2362	return Option.some(entry.exposedRuntime);
2363	}).pipe(
2364	Effect.mapError(
2365	(cause) =>
2366	new ProviderSessionLookupError({
2367	providerSessionId,
2368	cause,
2369	}),
2370	),
2371	),
2372	close: (providerSessionId) =>
2373	releaseEntry({ providerSessionId, reason: "manual_shutdown" }).pipe(
2374	Effect.mapError(
2375	(cause) =>
2376	new ProviderSessionCloseError({
2377	providerSessionId,
2378	cause,
2379	}),
2380	),
2381	),
2382	closeInstance: (instanceId) =>
2383	Effect.gen(function* () {
2384	const active = [...(yield* Ref.get(sessions)).values()].filter(
2385	(entry) => entry.runtime.instanceId === instanceId,
2386	);
2387	const outcomes = yield* Effect.forEach(
2388	active,
2389	(entry) =>
2390	releaseEntry({
2391	providerSessionId: entry.runtime.providerSessionId,
2392	reason: "manual_shutdown",
2393	detail: `Provider instance ${instanceId} logged out.`,
2394	}).pipe(Effect.exit),
2395	{ concurrency: "unbounded" },
2396	);
2397	const failure = outcomes.find(Exit.isFailure);
2398	if (failure !== undefined && Exit.isFailure(failure)) {
2399	return yield* Effect.failCause(failure.cause);
2400	}
2401	}).pipe(
2402	Effect.mapError(
2403	(cause) =>
2404	new ProviderSessionCloseError({
2405	providerSessionId: ProviderSessionId.make(
2406	`provider-session:provider-instance:${instanceId}`,
2407	),
2408	cause,
2409	}),
2410	),
2411	),
2412	release: releaseEntry,
2413	detach: (input) =>
2414	Effect.gen(function* () {
2415	const key = sessionKey(input.providerSessionId);
2416	const currentEntry = (yield* Ref.get(sessions)).get(key);
2417	let detachedProviderThreads: ReadonlyArray<OrchestrationV2ProviderThread> = [];
2418	if (currentEntry?.supportsMultipleProviderThreads === true) {
2419	const projection = yield* Effect.option(
2420	projectionStore.getThreadRecords(input.threadId, [
2421	"providerThreads",
2422	"providerTurns",
2423	]),
2424	);
2425	if (Option.isSome(projection)) {
2426	const providerThreads = new Map(
2427	projection.value.providerThreads
2428	.filter((thread) => thread.providerSessionId === input.providerSessionId)
2429	.map((thread) => [thread.id, thread] as const),
2430	);
2431	detachedProviderThreads = [...providerThreads.values()];
2432	const activeTurns = projection.value.providerTurns.filter(
2433	(turn) => turn.status === "running" && providerThreads.has(turn.providerThreadId),
2434	);
2435	yield* Effect.forEach(
2436	activeTurns,
2437	(turn) =>
2438	currentEntry.exposedRuntime
2439	.interruptTurn({
2440	providerThread: providerThreads.get(turn.providerThreadId)!,
2441	providerTurnId: turn.id,
2442	})
2443	.pipe(
2444	Effect.catchCause((cause) =>
2445	Effect.logWarning(
2446	"orchestration-v2.driver-session.detach-interrupt-failed",
2447	{
2448	providerSessionId: input.providerSessionId,
2449	threadId: input.threadId,
2450	providerTurnId: turn.id,
2451	cause,
2452	},
2453	),
2454	),
2455	),
2456	{ concurrency: 1, discard: true },
2457	);
2458	}
2459	}
2460	const detachResult = yield* Ref.modify(sessions, (current) => {
2461	const entry = current.get(key);
2462	if (entry === undefined || !entry.attachedThreadIds.has(input.threadId)) {
2463	return [
2464	Option.none<{
2465	readonly entry: LiveSessionEntry;
2466	readonly idleUnloadFiber: Fiber.Fiber<void, never> | null;
2467	}>(),
2468	current,
2469	] as const;
2470	}
2471	const attachedThreadIds = new Set(entry.attachedThreadIds);
2472	attachedThreadIds.delete(input.threadId);
2473	const loadedProviderThreadKeyByThread = new Map(
2474	entry.loadedProviderThreadKeyByThread,
2475	);
2476	loadedProviderThreadKeyByThread.delete(input.threadId);
2477	// For a plain (workspace-change) detach, the credential id stays
2478	// recorded: the thread may re-attach and reuse it, and
2479	// releaseEntry revokes it when the provider process finally goes
2480	// away. A terminal detach (archive/delete) prunes the record so
2481	// nothing vetoes the revocation below.
2482	const mcpCredentialIdByThread =
2483	input.revokeMcpCredential === true
2484	? (() => {
2485	const pruned = new Map(entry.mcpCredentialIdByThread);
2486	pruned.delete(input.threadId);
2487	return pruned;
2488	})()
2489	: entry.mcpCredentialIdByThread;
2490	// The detach unloads the thread itself below.
2491	const idleThreadUnloads = new Map(entry.idleThreadUnloads);
2492	idleThreadUnloads.delete(input.threadId);
2493	const updatedEntry = {
2494	...entry,
2495	attachedThreadIds,
2496	loadedProviderThreadKeyByThread,
2497	mcpCredentialIdByThread,
2498	idleThreadUnloads,
2499	};
2500	const updated = new Map(current);
2501	updated.set(key, updatedEntry);
2502	return [
2503	Option.some({
2504	entry: updatedEntry,
2505	idleUnloadFiber: entry.idleThreadUnloads.get(input.threadId)?.fiber ?? null,
2506	}),
2507	updated,
2508	] as const;
2509	});
2510	if (Option.isSome(detachResult)) {
2511	yield* cancelIdleFiber(detachResult.value.idleUnloadFiber);
2512	}
2513	const detached = Option.map(detachResult, (result) => result.entry);
2514	// Plain detaches deliberately do not revoke: a detached thread's
2515	// provider process may still be alive (shared multi-thread codex
2516	// session across a workspace handoff) and holds its MCP client's
2517	// credential for the thread it will re-attach with. Credentials
2518	// are revoked when the session entry is released (process gone)
2519	// or rotated on the next attach if they stopped resolving.
2520	// Terminal detaches (thread archived or deleted) revoke the
2521	// thread's credentials immediately, even on a retry where the
2522	// entry is already gone: there is no legitimate future re-attach,
2523	// and the token must not outlive the thread.
2524	if (input.revokeMcpCredential === true) {
2525	yield* clearMcpSession(input.threadId);
2526	}
2527	if (Option.isNone(detached)) {
2528	return;
2529	}
2530	if (
2531	detached.value.attachedThreadIds.size === 0 &&
2532	!detached.value.supportsMultipleProviderThreads
2533	) {
2534	yield* releaseEntry({
2535	providerSessionId: input.providerSessionId,
2536	reason: "manual_shutdown",
2537	...(input.detail === undefined ? {} : { detail: input.detail }),
2538	detachedThreadId: input.threadId,
2539	});
2540	return;
2541	}
2542	// The shared runtime stays up for other threads, so unload this
2543	// thread's native state rather than leaving it (and its MCP
2544	// servers) resident until the whole runtime is released.
2545	const unloadThread = detached.value.exposedRuntime.unloadThread;
2546	if (detached.value.supportsMultipleProviderThreads && unloadThread !== undefined) {
2547	// Serialized with re-attachment: a thread whose next turn
2548	// attaches first stays loaded, and one that attaches during the
2549	// unload waits for it, so its resume reloads the native thread.
2550	yield* threadAttachment.withLock(
2551	threadAttachmentKey(input),
2552	Effect.gen(function* () {
2553	const entry = (yield* Ref.get(sessions)).get(key);
2554	if (
2555	entry?.runtime !== detached.value.runtime ||
2556	entry.attachedThreadIds.has(input.threadId)
2557	) {
2558	return;
2559	}
2560	yield* Effect.forEach(
2561	detachedProviderThreads.filter((thread) => thread.nativeThreadRef !== null),
2562	(providerThread) =>
2563	unloadThread({ providerThread }).pipe(
2564	// Bounded so a wedged provider cannot hold up the
2565	// thread's next attach.
2566	Effect.timeout(UNLOAD_THREAD_TIMEOUT_MS),
2567	Effect.catchCause((cause) =>
2568	Effect.logWarning(
2569	"orchestration-v2.driver-session.detach-unload-failed",
2570	{
2571	providerSessionId: input.providerSessionId,
2572	threadId: input.threadId,
2573	providerThreadId: providerThread.id,
2574	cause,
2575	},
2576	),
2577	),
2578	),
2579	{ concurrency: 1, discard: true },
2580	);
2581	}),
2582	);
2583	}
2584	yield* scheduleIdleRelease(input.providerSessionId);
2585	}).pipe(
2586	Effect.catchCause((cause) =>
2587	Effect.fail(
2588	new ProviderSessionReleaseError({
2589	providerSessionId: input.providerSessionId,
2590	reason: "manual_shutdown",
2591	cause,
2592	}),
2593	),
2594	),
2595	),
2596	} satisfies ProviderSessionManagerV2Shape);
2597	}),
2598	);
2599
2600	export const layer = layerWithOptions();