diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index e8e4bf37d4794..059106a2040e5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -4660,6 +4660,57 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + // regression: #30630 Codex replies to unsubscribe before it closes the thread and frees its writer. + it.effect("unloading a thread waits for Codex to close it", () => + Effect.gen(function* () { + const transcript = makeCodexReplayTranscript({ + scenario: "codex-unload-waits-for-thread-closed", + entries: [ + ...backgroundExecTranscript.entries.slice(0, -2), + backgroundExecTranscript.entries.at(-2)!, + { + type: "expect_outbound", + label: "unsubscribe", + frame: { id: 4, method: "thread/unsubscribe", params: { threadId: BG_NATIVE_THREAD } }, + }, + { + type: "emit_inbound", + label: "unsubscribe-result", + frame: { id: 4, result: { status: "unsubscribed" } }, + }, + { + type: "emit_inbound", + label: "thread/closed", + frame: { method: "thread/closed", params: { threadId: BG_NATIVE_THREAD } }, + }, + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-codex-unload-waits-for-close"), + text: BG_PROMPT, + }), + ); + yield* harness.firstTerminal; + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const backgroundStopCases = [true, false, "still_running", "thread_unloaded"] as const; const makeBackgroundStopTranscript = (terminated: (typeof backgroundStopCases)[number]) => { const stillRunning = terminated === "still_running"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 66bc21d6030f2..0a81094208ead 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1940,6 +1940,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( string, { readonly runtimeMode: RuntimeMode; readonly kinds: Set } >(); + const threadCloses = new Map>>(); /** Bind answers to the request's approval set so mode changes invalidate late answers. */ const endSessionApprovalsOnModeChange = (context: ActiveCodexTurnContext) => { const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId; @@ -4979,7 +4980,17 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); yield* releaseSettledTurnIfIdle(context.nativeTurnId); } - }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)), + }).pipe( + Effect.orDie, + turnTerminalizationPermit.withPermits(1), + Effect.andThen(() => + Effect.forEach( + threadCloses.get(payload.threadId) ?? [], + (closed) => Deferred.succeed(closed, undefined), + { discard: true }, + ), + ), + ), ); yield* client.handleServerNotification("item/started", (payload) => @@ -7501,7 +7512,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( function* (unloadInput) { const nativeThreadId = yield* getNativeThreadId(unloadInput.providerThread); sessionApprovals.delete(nativeThreadId); - yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); + // Codex replies before it closes the thread; the writer lock and a same-process + // resume ("is closing") both wait for thread/closed. Callers bound the wait. + const closed = yield* Deferred.make(); + const waiters = threadCloses.get(nativeThreadId) ?? new Set(); + waiters.add(closed); + threadCloses.set(nativeThreadId, waiters); + yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }).pipe( + Effect.flatMap(({ status }) => + status === "unsubscribed" ? Deferred.await(closed) : Effect.void, + ), + Effect.ensuring( + Effect.sync(() => { + waiters.delete(closed); + if (waiters.size === 0 && threadCloses.get(nativeThreadId) === waiters) { + threadCloses.delete(nativeThreadId); + } + }), + ), + ); }, (effect, unloadInput) => effect.pipe( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b4fc69727c1de..37c6d06960cd9 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3790,6 +3790,7 @@ it.effect( threadId: secondThreadId, providerSessionId, now, + nativeThreadId: "native-thread-second", }); const firstRunId = idAllocator.derive.run({ threadId: firstThreadId, ordinal: 1 }); yield* eventSink.write({ @@ -3863,7 +3864,7 @@ it.effect( yield* settings.updateSettings({ linkedEnvironmentsEnabled: true }); yield* resumeSecondThread; assert.equal((yield* Ref.get(state)).resumeCount, 4); - assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]); + assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread-second"]); const subscribe = firstRuntime.subscribeEvents; assert.isDefined(subscribe); if (subscribe === undefined) return; @@ -3897,14 +3898,18 @@ it.effect( yield* resumeSecondThread; assert.equal((yield* Ref.get(state)).resumeCount, 5); - assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]); + assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [ + "native-thread-second", + "native-thread-second", + ]); yield* manager.detach({ providerSessionId, threadId: firstThreadId }); assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); assert.equal((yield* Ref.get(state)).closeCount, 0); assert.equal((yield* Ref.get(state)).interruptCount, 1); assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [ - "native-thread", + "native-thread-second", + "native-thread-second", "native-thread", ]); @@ -4228,6 +4233,74 @@ it.effect( }), ); +it.effect( + "ProviderSessionManagerV2 unloads a thread from its old session after it resumes on another instance", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("cross-instance-switch"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + const providerThread = makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: providerThread, + }, + ], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* runtime.resumeThread({ providerThread, threadId, modelSelection, runtimePolicy }); + + // A restart_and_resume switch moves the same row to the other instance's session. + const otherInstanceId = ProviderInstanceId.make("codex-other"); + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload: { + ...providerThread, + providerInstanceId: otherInstanceId, + providerSessionId: idAllocator.derive.providerSession({ + providerInstanceId: otherInstanceId, + }), + }, + }, + ], + }); + yield* manager.detach({ providerSessionId, threadId }); + + assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })), Effect.scoped); + }), +); + it.effect( "ProviderSessionManagerV2 re-attaching a thread waits for its in-flight unload, then reloads it", () => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 0fb31e81a0732..2e233492eb8a2 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2726,6 +2726,7 @@ export const layerWithOptions = ( return [ Option.none<{ readonly entry: LiveSessionEntry; + readonly heldProviderThread: OrchestrationV2ProviderThread | undefined; readonly idleUnloadFiber: Fiber.Fiber | null; }>(), current, @@ -2762,6 +2763,7 @@ export const layerWithOptions = ( return [ Option.some({ entry: updatedEntry, + heldProviderThread: entry.idleThreadUnloads.get(input.threadId)?.providerThread, idleUnloadFiber: entry.idleThreadUnloads.get(input.threadId)?.fiber ?? null, }), updated, @@ -2770,17 +2772,17 @@ export const layerWithOptions = ( if (Option.isSome(detachResult)) { yield* cancelIdleFiber(detachResult.value.idleUnloadFiber); } - const detached = Option.map(detachResult, (result) => result.entry); // Plain detaches preserve credentials held by live shared providers across workspace handoffs. if (input.revokeMcpCredential === true) { yield* clearMcpSession(input.threadId); } - if (Option.isNone(detached)) { + if (Option.isNone(detachResult)) { return; } + const { entry: detached, heldProviderThread } = detachResult.value; if ( - detached.value.attachedThreadIds.size === 0 && - !detached.value.supportsMultipleProviderThreads + detached.attachedThreadIds.size === 0 && + !detached.supportsMultipleProviderThreads ) { yield* releaseEntry({ providerSessionId: input.providerSessionId, @@ -2790,8 +2792,16 @@ export const layerWithOptions = ( }); return; } - const unloadThread = detached.value.exposedRuntime.unloadThread; - if (detached.value.supportsMultipleProviderThreads && unloadThread !== undefined) { + const unloadThread = detached.exposedRuntime.unloadThread; + if (detached.supportsMultipleProviderThreads && unloadThread !== undefined) { + // A cross-instance switch repoints the row before this detach, so the projection + // can miss what this session still holds for the thread. + if ( + heldProviderThread !== undefined && + !detachedProviderThreads.some((thread) => thread.id === heldProviderThread.id) + ) { + detachedProviderThreads = [...detachedProviderThreads, heldProviderThread]; + } // The projection includes native children owned by other app threads; // lock each owner so an attached child stays loaded and later attaches wait. for (const providerThread of detachedProviderThreads) { @@ -2805,7 +2815,7 @@ export const layerWithOptions = ( Effect.gen(function* () { const entry = (yield* Ref.get(sessions)).get(key); if ( - entry?.runtime !== detached.value.runtime || + entry?.runtime !== detached.runtime || entry.attachedThreadIds.has(ownerThreadId) ) { return; diff --git a/apps/server/src/provider/codexLaunchArgs.test.ts b/apps/server/src/provider/codexLaunchArgs.test.ts index ce18c8f3f1bd5..e5479d871010a 100644 --- a/apps/server/src/provider/codexLaunchArgs.test.ts +++ b/apps/server/src/provider/codexLaunchArgs.test.ts @@ -11,6 +11,16 @@ describe("codexAppServerArgs", () => { ["app-server", "--config", 'model="gpt-6.1-sol"', "--strict-config"], ); }); + + it("unloads unsubscribed threads immediately unless strict config could reject the key", () => { + NodeAssert.deepStrictEqual(codexAppServerArgs("--enable foo"), [ + "app-server", + "--config", + "thread_unload_delay_secs=0", + "--enable", + "foo", + ]); + }); }); describe("codexExecLaunchArgs", () => { diff --git a/apps/server/src/provider/codexLaunchArgs.ts b/apps/server/src/provider/codexLaunchArgs.ts index 33def5ce9a61a..a31ea1dd6aefa 100644 --- a/apps/server/src/provider/codexLaunchArgs.ts +++ b/apps/server/src/provider/codexLaunchArgs.ts @@ -9,10 +9,17 @@ export const resolveCodexLaunchArgs = ( const codexLaunchArgv = (launchArgs?: string): ReadonlyArray => tokenizeCliArgs(launchArgs); -export const codexAppServerArgs = (launchArgs?: string) => [ - "app-server", - ...codexLaunchArgv(launchArgs), -]; +export const codexAppServerArgs = (launchArgs?: string) => { + const args = codexLaunchArgv(launchArgs); + return [ + "app-server", + // Otherwise Codex keeps an unsubscribed thread, and its writer lock, loaded for a delay. + // Codex 0.163 ignores this key in per-thread config, so it is set for the app-server. + // Codex before 0.154 lacks the key, which --strict-config rejects at startup. + ...(args.includes("--strict-config") ? [] : ["--config", "thread_unload_delay_secs=0"]), + ...args, + ]; +}; export const codexExecLaunchArgs = (launchArgs?: string) => { const args = codexLaunchArgv(launchArgs);