authorgravatar for 39114868+RhysSullivan@users.noreply.github.comRhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> 2026-10-07 13:54:17-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 20:54:17+00:00
logeba0521563946a0834af881465d3e1ff7a4ca2f1
tree0676d482356e5fbf6f423b284033987906ce3b12
parent85489611b5477050d7d777a4195c1c50ac0ce6be
signature Signed by PGP key B5690EEEBB952194

fix(server): unload Codex threads left idle on the shared app-server (#16917)

Co-authored-by: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

2 files changed, 440 insertions(+), 11 deletions(-)

apps/server/src/orchestration-v2/ProviderSessionManager.test.ts+207
...@@ -319,6 +319,7 @@ function makeProviderAdapter(...@@ -319,6 +319,7 @@ function makeProviderAdapter(
319 readonly initialProviderItemIdentityVersion?: 2;319 readonly initialProviderItemIdentityVersion?: 2;
320 }) => Effect.Effect<void>;320 }) => Effect.Effect<void>;
321 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;321 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
322 readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
322 readonly hangSessionScopeClose?: boolean;323 readonly hangSessionScopeClose?: boolean;
323 readonly startTurn?: Effect.Effect<void>;324 readonly startTurn?: Effect.Effect<void>;
324 readonly beforeUnload?: Effect.Effect<void>;325 readonly beforeUnload?: Effect.Effect<void>;
...@@ -407,6 +408,11 @@ function makeProviderAdapter(...@@ -407,6 +408,11 @@ function makeProviderAdapter(
407 ...(options.hasPendingBackgroundWork === undefined408 ...(options.hasPendingBackgroundWork === undefined
408 ? {}409 ? {}
409 : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }),410 : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }),
411 ...(options.hasPendingBackgroundWorkForThread === undefined
412 ? {}
413 : {
414 hasPendingBackgroundWorkForThread: () => options.hasPendingBackgroundWorkForThread!,
415 }),
410 ensureThread: () => unimplemented("ensureThread unused in test"),416 ensureThread: () => unimplemented("ensureThread unused in test"),
411 resumeThread: (threadInput) =>417 resumeThread: (threadInput) =>
412 Ref.update(state, (current) => ({418 Ref.update(state, (current) => ({
...@@ -463,6 +469,7 @@ function layerTest(input: {...@@ -463,6 +469,7 @@ function layerTest(input: {
463 readonly paused: Deferred.Deferred<void>;469 readonly paused: Deferred.Deferred<void>;
464 };470 };
465 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;471 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
472 readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
466 readonly hangSessionScopeClose?: boolean;473 readonly hangSessionScopeClose?: boolean;
467 readonly startTurn?: Effect.Effect<void>;474 readonly startTurn?: Effect.Effect<void>;
468 readonly beforeUnload?: Effect.Effect<void>;475 readonly beforeUnload?: Effect.Effect<void>;
...@@ -488,6 +495,9 @@ function layerTest(input: {...@@ -488,6 +495,9 @@ function layerTest(input: {
488 ...(input.hasPendingBackgroundWork === undefined495 ...(input.hasPendingBackgroundWork === undefined
489 ? {}496 ? {}
490 : { hasPendingBackgroundWork: input.hasPendingBackgroundWork }),497 : { hasPendingBackgroundWork: input.hasPendingBackgroundWork }),
498 ...(input.hasPendingBackgroundWorkForThread === undefined
499 ? {}
500 : { hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread }),
491 ...(input.hangSessionScopeClose === undefined501 ...(input.hangSessionScopeClose === undefined
492 ? {}502 ? {}
493 : { hangSessionScopeClose: input.hangSessionScopeClose }),503 : { hangSessionScopeClose: input.hangSessionScopeClose }),
...@@ -4210,6 +4220,203 @@ it.effect(...@@ -4210,6 +4220,203 @@ it.effect(
4210 }),4220 }),
4211);4221);
42124222
4223function runIdleThreadUnloadScenario(
4224 name: string,
4225 scenario: (input: {
4226 readonly state: Ref.Ref<TestProviderRuntimeState>;
4227 readonly manager: ProviderSessionManager.ProviderSessionManagerV2Shape;
4228 readonly providerSessionId: ProviderSessionId;
4229 readonly threadA: ThreadId;
4230 /** Starts run `ordinal` on a thread and returns once the provider accepted it. */
4231 readonly startTurn: (threadId: ThreadId, ordinal: number) => Effect.Effect<void>;
4232 /** Ends run `ordinal` on a thread and waits for the session to process it. */
4233 readonly endTurn: (threadId: ThreadId, ordinal: number) => Effect.Effect<void>;
4234 readonly resume: (threadId: ThreadId) => Effect.Effect<void>;
4235 }) => Effect.Effect<void, ProviderSessionManager.ProviderSessionManagerV2Error>,
4236 options: { readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean> } = {},
4237) {
4238 return Effect.gen(function* () {
4239 const state = yield* Ref.make(emptyState);
4240 yield* Effect.gen(function* () {
4241 const eventSink = yield* EventSink.EventSinkV2;
4242 const idAllocator = yield* IdAllocator.IdAllocatorV2;
4243 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4244 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4245 const now = yield* DateTime.now;
4246 const threadA = ThreadId.make(`thread-provider-session-manager-${name}-a`);
4247 const threadB = ThreadId.make(`thread-provider-session-manager-${name}-b`);
4248 const providerSessionId = idAllocator.derive.providerSession({
4249 providerInstanceId: modelSelection.instanceId,
4250 });
4251 yield* eventSink.write({
4252 events: [
4253 yield* makeThreadCreatedEvent({ idAllocator, threadId: threadA, now }),
4254 yield* makeThreadCreatedEvent({ idAllocator, threadId: threadB, now }),
4255 ],
4256 });
4257 const runtime = yield* manager.open({
4258 threadId: threadA,
4259 providerSessionId,
4260 modelSelection,
4261 runtimePolicy,
4262 });
4263 yield* manager.open({ threadId: threadB, providerSessionId, modelSelection, runtimePolicy });
4264 const providerThreadOf = (threadId: ThreadId) =>
4265 makeProviderThread({
4266 idAllocator,
4267 threadId,
4268 providerSessionId,
4269 now,
4270 nativeThreadId: `native-${threadId}`,
4271 });
4272 const resume = (threadId: ThreadId) =>
4273 runtime
4274 .resumeThread({
4275 providerThread: providerThreadOf(threadId),
4276 threadId,
4277 modelSelection,
4278 runtimePolicy,
4279 })
4280 .pipe(Effect.asVoid, Effect.orDie);
4281 const startTurn = (threadId: ThreadId, ordinal: number) =>
4282 Effect.gen(function* () {
4283 yield* resume(threadId);
4284 const runId = idAllocator.derive.run({ threadId, ordinal });
4285 yield* runtime.startTurn({
4286 appThread: (yield* projectionStore.getThreadProjection(threadId)).thread,
4287 threadId,
4288 runId,
4289 runOrdinal: ordinal,
4290 providerTurnOrdinal: ordinal,
4291 attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }),
4292 rootNodeId: idAllocator.derive.rootNode({ runId }),
4293 providerThread: providerThreadOf(threadId),
4294 message: {
4295 createdBy: "user",
4296 creationSource: "web",
4297 messageId: yield* idAllocator.allocate.message({ threadId, ordinal }),
4298 text: "turn",
4299 attachments: [],
4300 },
4301 modelSelection,
4302 runtimePolicy,
4303 });
4304 }).pipe(Effect.orDie);
4305 const subscribe = runtime.subscribeEvents;
4306 assert.isDefined(subscribe);
4307 const endTurn = (threadId: ThreadId, ordinal: number) =>
4308 Effect.scoped(
4309 Effect.gen(function* () {
4310 // The pump hands an event to subscribers only after the session
4311 // has processed it, so receiving it is the receipt.
4312 const subscription = yield* Effect.acquireRelease(subscribe!, (sub) => sub.close);
4313 const received = yield* subscription.events.pipe(
4314 Stream.filter((event) => event.type === "turn.terminal"),
4315 Stream.runHead,
4316 Effect.forkScoped,
4317 );
4318 const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
4319 assert.isDefined(queue);
4320 yield* Queue.offer(queue!, {
4321 type: "turn.terminal",
4322 driver: CODEX_DRIVER,
4323 providerThreadId: providerThreadOf(threadId).id,
4324 providerTurnId: idAllocator.derive.providerTurn({
4325 driver: CODEX_DRIVER,
4326 nativeTurnId: `native-turn-${threadId}-${ordinal}`,
4327 }),
4328 runOrdinal: ordinal,
4329 status: "completed",
4330 failure: null,
4331 threadDisposition: "reusable",
4332 });
4333 yield* Fiber.join(received);
4334 }),
4335 ).pipe(Effect.orDie);
4336
4337 // B's running turn keeps the shared runtime itself busy throughout.
4338 yield* startTurn(threadB, 1);
4339 yield* scenario({
4340 state,
4341 manager,
4342 providerSessionId,
4343 threadA,
4344 startTurn,
4345 endTurn,
4346 resume,
4347 });
4348 }).pipe(
4349 Effect.provide(
4350 layerTest({
4351 state,
4352 idleTimeoutMs: 1000,
4353 ...(options.hasPendingBackgroundWorkForThread === undefined
4354 ? {}
4355 : { hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread }),
4356 }),
4357 ),
4358 );
4359 });
4360}
4361
4362it.effect(
4363 "ProviderSessionManagerV2 unloads a shared-runtime thread left idle, and reloads it on its next turn",
4364 () =>
4365 runIdleThreadUnloadScenario(
4366 "idle-unload",
4367 ({ state, manager, providerSessionId, threadA, startTurn, endTurn, resume }) =>
4368 Effect.gen(function* () {
4369 yield* startTurn(threadA, 1);
4370 assert.equal((yield* Ref.get(state)).resumeCount, 2);
4371 yield* endTurn(threadA, 1);
4372
4373 // A follow-up before the timeout keeps the thread loaded.
4374 yield* TestClock.adjust("500 millis");
4375 yield* startTurn(threadA, 2);
4376 yield* TestClock.adjust("1 second");
4377 assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, []);
4378
4379 // Idle for the full timeout after its last turn, A is unloaded while
4380 // the runtime stays up for B.
4381 yield* endTurn(threadA, 2);
4382 yield* TestClock.adjust("1 second");
4383 assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [`native-${threadA}`]);
4384 assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
4385 assert.equal((yield* Ref.get(state)).closeCount, 0);
4386
4387 // The next turn's resume reaches the provider and reloads A.
4388 const resumes = (yield* Ref.get(state)).resumeCount;
4389 yield* resume(threadA);
4390 assert.equal((yield* Ref.get(state)).resumeCount, resumes + 1);
4391 }),
4392 ),
4393);
4394
4395it.effect(
4396 "ProviderSessionManagerV2 keeps an idle shared-runtime thread loaded while its background work runs",
4397 () =>
4398 Effect.gen(function* () {
4399 const pendingWork = yield* Ref.make(true);
4400 yield* runIdleThreadUnloadScenario(
4401 "idle-unload-pinned",
4402 ({ state, threadA, startTurn, endTurn }) =>
4403 Effect.gen(function* () {
4404 yield* startTurn(threadA, 1);
4405 yield* endTurn(threadA, 1);
4406 yield* TestClock.adjust("3 seconds");
4407 assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, []);
4408
4409 yield* Ref.set(pendingWork, false);
4410 yield* TestClock.adjust("1 second");
4411 assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [
4412 `native-${threadA}`,
4413 ]);
4414 }),
4415 { hasPendingBackgroundWorkForThread: Ref.get(pendingWork) },
4416 );
4417 }),
4418);
4419
4213it.effect(4420it.effect(
4214 "ProviderSessionManagerV2 rejects a second thread when the provider runtime is exclusive",4421 "ProviderSessionManagerV2 rejects a second thread when the provider runtime is exclusive",
4215 () =>4422 () =>
apps/server/src/orchestration-v2/ProviderSessionManager.ts+233-11
...@@ -61,9 +61,10 @@ const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000;...@@ -61,9 +61,10 @@ const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000;
61const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000;61const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000;
62const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000;62const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000;
6363
64const busyTurnPrefix = (providerThreadId: ProviderThreadId) => `${providerThreadId}#`;
64/** The identity a turn's start and its `turn.terminal` share. */65/** The identity a turn's start and its `turn.terminal` share. */
65const busyTurnKey = (providerThreadId: ProviderThreadId, runOrdinal: number) =>66const busyTurnKey = (providerThreadId: ProviderThreadId, runOrdinal: number) =>
66 `${providerThreadId}#${runOrdinal}`;67 `${busyTurnPrefix(providerThreadId)}${runOrdinal}`;
67const UNLOAD_THREAD_TIMEOUT_MS = 10 * 1000;68const UNLOAD_THREAD_TIMEOUT_MS = 10 * 1000;
6869
69export const ProviderSessionReleaseReason = Schema.Literals([70export const ProviderSessionReleaseReason = Schema.Literals([
...@@ -228,6 +229,18 @@ interface LiveSessionEntry {...@@ -228,6 +229,18 @@ interface LiveSessionEntry {
228 readonly idleFiber: Fiber.Fiber<void, never> | null;229 readonly idleFiber: Fiber.Fiber<void, never> | null;
229 /** Set when idle release is deferred for pending background work; bounds total deferral. */230 /** Set when idle release is deferred for pending background work; bounds total deferral. */
230 readonly pinnedSinceMs: number | null;231 readonly pinnedSinceMs: number | null;
232 /**
233 * Shared runtimes only: the provider thread each attached app thread last
234 * started a turn on, and the timer that unloads it once it has been idle
235 * for `idleTimeoutMs`.
236 */
237 readonly idleThreadUnloads: ReadonlyMap<ThreadId, IdleThreadUnload>;
238}
239
240interface IdleThreadUnload {
241 readonly providerThread: OrchestrationV2ProviderThread;
242 readonly generation: number;
243 readonly fiber: Fiber.Fiber<void, never> | null;
231}244}
232245
233type ProviderSessionEventSignal =246type ProviderSessionEventSignal =
...@@ -1400,7 +1413,11 @@ export const layerWithOptions = (...@@ -1400,7 +1413,11 @@ export const layerWithOptions = (
1400 // Clearing a turn that is not marked busy (one whose failed start already1413 // Clearing a turn that is not marked busy (one whose failed start already
1401 // cleared it, or a subagent turn the manager never started) only1414 // cleared it, or a subagent turn the manager never started) only
1402 // records activity.1415 // records activity.
1403 const markIdle = (providerSessionId: ProviderSessionId, turnKey: string) =>1416 const markIdle = (
1417 providerSessionId: ProviderSessionId,
1418 providerThreadId: ProviderThreadId,
1419 runOrdinal: number,
1420 ) =>
1404 withActivityError(1421 withActivityError(
1405 providerSessionId,1422 providerSessionId,
1406 Effect.gen(function* () {1423 Effect.gen(function* () {
...@@ -1412,7 +1429,7 @@ export const layerWithOptions = (...@@ -1412,7 +1429,7 @@ export const layerWithOptions = (
1412 return current;1429 return current;
1413 }1430 }
1414 const busyTurns = new Set(entry.busyTurns);1431 const busyTurns = new Set(entry.busyTurns);
1415 busyTurns.delete(turnKey);1432 busyTurns.delete(busyTurnKey(providerThreadId, runOrdinal));
1416 const updated = new Map(current);1433 const updated = new Map(current);
1417 updated.set(key, {1434 updated.set(key, {
1418 ...entry,1435 ...entry,
...@@ -1422,9 +1439,181 @@ export const layerWithOptions = (...@@ -1422,9 +1439,181 @@ export const layerWithOptions = (
1422 return updated;1439 return updated;
1423 });1440 });
1424 yield* scheduleIdleReleaseInternal(providerSessionId);1441 yield* scheduleIdleReleaseInternal(providerSessionId);
1442 yield* scheduleThreadUnload(providerSessionId, providerThreadId);
1425 }),1443 }),
1426 );1444 );
14271445
1446 const hasBusyTurn = (entry: LiveSessionEntry, providerThreadId: ProviderThreadId) => {
1447 const prefix = busyTurnPrefix(providerThreadId);
1448 for (const turnKey of entry.busyTurns) {
1449 if (turnKey.startsWith(prefix)) return true;
1450 }
1451 return false;
1452 };
1453
1454 const updateIdleThreadUnload = (
1455 providerSessionId: ProviderSessionId,
1456 threadId: ThreadId,
1457 update: (current: IdleThreadUnload | undefined) => IdleThreadUnload | undefined,
1458 ) =>
1459 Ref.modify(sessions, (current) => {
1460 const key = sessionKey(providerSessionId);
1461 const entry = current.get(key);
1462 if (entry === undefined) return [undefined, current] as const;
1463 const previous = entry.idleThreadUnloads.get(threadId);
1464 const next = update(previous);
1465 const idleThreadUnloads = new Map(entry.idleThreadUnloads);
1466 if (next === undefined) idleThreadUnloads.delete(threadId);
1467 else idleThreadUnloads.set(threadId, next);
1468 const updated = new Map(current);
1469 updated.set(key, { ...entry, idleThreadUnloads });
1470 return [previous, updated] as const;
1471 });
1472
1473 /**
1474 * Starts tracking the provider thread an app thread runs its turns on, and
1475 * stops any unload pending for it. Called before the thread is resumed or
1476 * given a turn, so an unload cannot land between a resume that found the
1477 * thread loaded and the turn that relies on it.
1478 */
1479 const holdThreadLoaded = (input: {
1480 readonly providerSessionId: ProviderSessionId;
1481 readonly threadId: ThreadId;
1482 readonly providerThread: OrchestrationV2ProviderThread;
1483 }) =>
1484 Effect.gen(function* () {
1485 const entry = (yield* Ref.get(sessions)).get(sessionKey(input.providerSessionId));
1486 if (
1487 entry === undefined ||
1488 !entry.supportsMultipleProviderThreads ||
1489 entry.exposedRuntime.unloadThread === undefined ||
1490 input.providerThread.nativeThreadRef === null
1491 ) {
1492 return;
1493 }
1494 const previous = yield* updateIdleThreadUnload(
1495 input.providerSessionId,
1496 input.threadId,
1497 (current) => ({
1498 providerThread: input.providerThread,
1499 generation: (current?.generation ?? 0) + 1,
1500 fiber: null,
1501 }),
1502 );
1503 yield* cancelIdleFiber(previous?.fiber ?? null);
1504 });
1505
1506 /**
1507 * A shared runtime never goes idle while any of its threads is in use, so
1508 * its threads get the idle timeout one by one: a thread with no turn for
1509 * `idleTimeoutMs` is unloaded from the runtime, along with the native MCP
1510 * servers it started. Its next turn's resume loads it again.
1511 */
1512 const scheduleThreadUnload = (
1513 providerSessionId: ProviderSessionId,
1514 providerThreadId: ProviderThreadId,
1515 ) =>
1516 Effect.gen(function* () {
1517 const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId));
1518 if (entry === undefined || hasBusyTurn(entry, providerThreadId)) return;
1519 const tracked = Array.from(entry.idleThreadUnloads).find(
1520 ([, pending]) => pending.providerThread.id === providerThreadId,
1521 );
1522 if (tracked === undefined) return;
1523 const [threadId, pending] = tracked;
1524 const generation = pending.generation + 1;
1525 const fiber = yield* Effect.sleep(Duration.millis(idleTimeoutMs)).pipe(
1526 Effect.andThen(unloadIdleThread({ providerSessionId, threadId, generation })),
1527 Effect.forkIn(layerScope),
1528 );
1529 // A turn that started meanwhile already moved the generation on.
1530 const previous = yield* updateIdleThreadUnload(providerSessionId, threadId, (current) =>
1531 current?.generation === pending.generation
1532 ? { ...current, generation, fiber }
1533 : current,
1534 );
1535 yield* cancelIdleFiber(
1536 previous?.generation === pending.generation ? previous.fiber : fiber,
1537 );
1538 });
1539
1540 const unloadIdleThread = (input: {
1541 readonly providerSessionId: ProviderSessionId;
1542 readonly threadId: ThreadId;
1543 readonly generation: number;
1544 }): Effect.Effect<void> =>
1545 Effect.gen(function* () {
1546 const outcome = yield* threadAttachment.withLock(
1547 threadAttachmentKey(input),
1548 Effect.gen(function* () {
1549 const key = sessionKey(input.providerSessionId);
1550 const entry = (yield* Ref.get(sessions)).get(key);
1551 const pending = entry?.idleThreadUnloads.get(input.threadId);
1552 const unloadThread = entry?.exposedRuntime.unloadThread;
1553 if (
1554 entry === undefined ||
1555 pending === undefined ||
1556 pending.generation !== input.generation ||
1557 unloadThread === undefined ||
1558 !entry.attachedThreadIds.has(input.threadId) ||
1559 hasBusyTurn(entry, pending.providerThread.id)
1560 ) {
1561 return "skipped" as const;
1562 }
1563 // Unloading stops the native thread's background terminals, so a
1564 // thread still running background work stays loaded.
1565 const hasPendingWork =
1566 entry.runtime.hasPendingBackgroundWorkForThread === undefined
1567 ? false
1568 : yield* entry.runtime
1569 .hasPendingBackgroundWorkForThread(pending.providerThread)
1570 .pipe(Effect.catchCause(() => Effect.succeed(false)));
1571 if (hasPendingWork) return "deferred" as const;
1572 const unloading = yield* Ref.modify(sessions, (current) => {
1573 const latest = current.get(key);
1574 if (
1575 latest?.runtime !== entry.runtime ||
1576 latest.idleThreadUnloads.get(input.threadId)?.generation !== input.generation
1577 ) {
1578 return [false, current] as const;
1579 }
1580 const loadedProviderThreadKeyByThread = new Map(
1581 latest.loadedProviderThreadKeyByThread,
1582 );
1583 loadedProviderThreadKeyByThread.delete(input.threadId);
1584 const idleThreadUnloads = new Map(latest.idleThreadUnloads);
1585 idleThreadUnloads.delete(input.threadId);
1586 const updated = new Map(current);
1587 updated.set(key, {
1588 ...latest,
1589 loadedProviderThreadKeyByThread,
1590 idleThreadUnloads,
1591 });
1592 return [true, updated] as const;
1593 });
1594 if (!unloading) return "skipped" as const;
1595 yield* unloadThread({ providerThread: pending.providerThread }).pipe(
1596 Effect.timeout(UNLOAD_THREAD_TIMEOUT_MS),
1597 Effect.catchCause((cause) =>
1598 Effect.logWarning("orchestration-v2.driver-session.idle-unload-failed", {
1599 providerSessionId: input.providerSessionId,
1600 threadId: input.threadId,
1601 providerThreadId: pending.providerThread.id,
1602 cause,
1603 }),
1604 ),
1605 );
1606 return "unloaded" as const;
1607 }),
1608 );
1609 // Re-check on this fiber after another idle window, outside the
1610 // lock so the thread's next attach is not held up meanwhile.
1611 if (outcome === "deferred") {
1612 yield* Effect.sleep(Duration.millis(idleTimeoutMs));
1613 return yield* unloadIdleThread(input);
1614 }
1615 });
1616
1428 const observeActivity = (1617 const observeActivity = (
1429 providerSessionId: ProviderSessionId,1618 providerSessionId: ProviderSessionId,
1430 activity: Effect.Effect<void, ProviderSessionActivityError>,1619 activity: Effect.Effect<void, ProviderSessionActivityError>,
...@@ -1544,6 +1733,13 @@ export const layerWithOptions = (...@@ -1544,6 +1733,13 @@ export const layerWithOptions = (
1544 providerInstanceId: runtime.instanceId,1733 providerInstanceId: runtime.instanceId,
1545 }),1734 }),
1546 ).pipe(1735 ).pipe(
1736 Effect.andThen(
1737 holdThreadLoaded({
1738 providerSessionId,
1739 threadId,
1740 providerThread: input.providerThread,
1741 }),
1742 ),
1547 Effect.andThen(1743 Effect.andThen(
1548 isProviderThreadLoaded({ providerSessionId, threadId, providerThreadKey }),1744 isProviderThreadLoaded({ providerSessionId, threadId, providerThreadKey }),
1549 ),1745 ),
...@@ -1607,6 +1803,13 @@ export const layerWithOptions = (...@@ -1607,6 +1803,13 @@ export const layerWithOptions = (
1607 // the adapter emits the terminal anyway, clearing the same turn1803 // the adapter emits the terminal anyway, clearing the same turn
1608 // again changes nothing, so another thread's turn on a shared1804 // again changes nothing, so another thread's turn on a shared
1609 // session stays busy either way.1805 // session stays busy either way.
1806 Effect.andThen(
1807 holdThreadLoaded({
1808 providerSessionId,
1809 threadId: input.threadId,
1810 providerThread: input.providerThread,
1811 }),
1812 ),
1610 Effect.andThen(1813 Effect.andThen(
1611 Effect.acquireUseRelease(1814 Effect.acquireUseRelease(
1612 observeActivity(1815 observeActivity(
...@@ -1622,10 +1825,7 @@ export const layerWithOptions = (...@@ -1622,10 +1825,7 @@ export const layerWithOptions = (
1622 Exit.isFailure(exit)1825 Exit.isFailure(exit)
1623 ? observeActivity(1826 ? observeActivity(
1624 providerSessionId,1827 providerSessionId,
1625 markIdle(1828 markIdle(providerSessionId, input.providerThread.id, input.runOrdinal),
1626 providerSessionId,
1627 busyTurnKey(input.providerThread.id, input.runOrdinal),
1628 ),
1629 )1829 )
1630 : Effect.void,1830 : Effect.void,
1631 ),1831 ),
...@@ -1692,7 +1892,8 @@ export const layerWithOptions = (...@@ -1692,7 +1892,8 @@ export const layerWithOptions = (
1692 event.type === "turn.terminal"1892 event.type === "turn.terminal"
1693 ? markIdle(1893 ? markIdle(
1694 entry.runtime.providerSessionId,1894 entry.runtime.providerSessionId,
1695 busyTurnKey(event.providerThreadId, event.runOrdinal),1895 event.providerThreadId,
1896 event.runOrdinal,
1696 )1897 )
1697 : touchActivity(entry.runtime.providerSessionId),1898 : touchActivity(entry.runtime.providerSessionId),
1698 ).pipe(1899 ).pipe(
...@@ -1965,6 +2166,7 @@ export const layerWithOptions = (...@@ -1965,6 +2166,7 @@ export const layerWithOptions = (
1965 lastActivityAtMs: now,2166 lastActivityAtMs: now,
1966 idleFiber: null,2167 idleFiber: null,
1967 pinnedSinceMs: null,2168 pinnedSinceMs: null,
2169 idleThreadUnloads: new Map(),
1968 };2170 };
1969 yield* Ref.update(sessions, (current) => {2171 yield* Ref.update(sessions, (current) => {
1970 const updated = new Map(current);2172 const updated = new Map(current);
...@@ -2111,10 +2313,16 @@ export const layerWithOptions = (...@@ -2111,10 +2313,16 @@ export const layerWithOptions = (
2111 );2313 );
2112 }2314 }
2113 }2315 }
2114 const detached = yield* Ref.modify(sessions, (current) => {2316 const detachResult = yield* Ref.modify(sessions, (current) => {
2115 const entry = current.get(key);2317 const entry = current.get(key);
2116 if (entry === undefined || !entry.attachedThreadIds.has(input.threadId)) {2318 if (entry === undefined || !entry.attachedThreadIds.has(input.threadId)) {
2117 return [Option.none<LiveSessionEntry>(), current] as const;2319 return [
2320 Option.none<{
2321 readonly entry: LiveSessionEntry;
2322 readonly idleUnloadFiber: Fiber.Fiber<void, never> | null;
2323 }>(),
2324 current,
2325 ] as const;
2118 }2326 }
2119 const attachedThreadIds = new Set(entry.attachedThreadIds);2327 const attachedThreadIds = new Set(entry.attachedThreadIds);
2120 attachedThreadIds.delete(input.threadId);2328 attachedThreadIds.delete(input.threadId);
...@@ -2135,16 +2343,30 @@ export const layerWithOptions = (...@@ -2135,16 +2343,30 @@ export const layerWithOptions = (
2135 return pruned;2343 return pruned;
2136 })()2344 })()
2137 : entry.mcpCredentialIdByThread;2345 : entry.mcpCredentialIdByThread;
2346 // The detach unloads the thread itself below.
2347 const idleThreadUnloads = new Map(entry.idleThreadUnloads);
2348 idleThreadUnloads.delete(input.threadId);
2138 const updatedEntry = {2349 const updatedEntry = {
2139 ...entry,2350 ...entry,
2140 attachedThreadIds,2351 attachedThreadIds,
2141 loadedProviderThreadKeyByThread,2352 loadedProviderThreadKeyByThread,
2142 mcpCredentialIdByThread,2353 mcpCredentialIdByThread,
2354 idleThreadUnloads,
2143 };2355 };
2144 const updated = new Map(current);2356 const updated = new Map(current);
2145 updated.set(key, updatedEntry);2357 updated.set(key, updatedEntry);
2146 return [Option.some(updatedEntry), updated] as const;2358 return [
2359 Option.some({
2360 entry: updatedEntry,
2361 idleUnloadFiber: entry.idleThreadUnloads.get(input.threadId)?.fiber ?? null,
2362 }),
2363 updated,
2364 ] as const;
2147 });2365 });
2366 if (Option.isSome(detachResult)) {
2367 yield* cancelIdleFiber(detachResult.value.idleUnloadFiber);
2368 }
2369 const detached = Option.map(detachResult, (result) => result.entry);
2148 // Plain detaches deliberately do not revoke: a detached thread's2370 // Plain detaches deliberately do not revoke: a detached thread's
2149 // provider process may still be alive (shared multi-thread codex2371 // provider process may still be alive (shared multi-thread codex
2150 // session across a workspace handoff) and holds its MCP client's2372 // session across a workspace handoff) and holds its MCP client's