diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 09ec3034a9bc9..64b72b1655aa8 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -3581,6 +3581,101 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + // regression: #30630 Codex answers thread/unsubscribe while it still holds the writer lock + it.effect( + "a detach unload waits for thread/closed, and only once when Codex never sends it", + () => + Effect.gen(function* () { + const nativeThreadId = "handed-over-thread"; + const unsubscribe = ( + id: number, + status = "unsubscribed", + ): ReadonlyArray => [ + { + type: "expect_outbound", + label: `unsubscribe-${id}`, + frame: { id, method: "thread/unsubscribe", params: { threadId: nativeThreadId } }, + }, + { + type: "emit_inbound", + label: `unsubscribe-result-${id}`, + frame: { id, result: { status } }, + }, + ]; + const closedFrame = (threadId: string): CodexReplay.CodexAppServerReplayEntry => ({ + type: "emit_inbound", + label: `closed-${threadId}`, + frame: { method: "thread/closed", params: { threadId } }, + }); + const closedNotifications = yield* Queue.unbounded(); + const closeReached = yield* Deferred.make(); + const releaseClose = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "codex-detach-unload-awaits-closed", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }).slice(0, 5), + // An idle unload leaves the close pending; Codex then answers notSubscribed. + ...unsubscribe(3), + ...unsubscribe(4, "notSubscribed"), + closedFrame(nativeThreadId), + ...unsubscribe(5), + // Replies fence notifications on the protocol reader, so this one marks reply 5 as read. + closedFrame("unrelated-thread"), + ...unsubscribe(6), + ], + }), + undefined, + undefined, + undefined, + (entry) => + entry.label === `closed-${nativeThreadId}` + ? Deferred.succeed(closeReached, undefined).pipe( + Effect.andThen(Deferred.await(releaseClose)), + ) + : Effect.void, + undefined, + undefined, + (method) => + method === "thread/closed" ? Queue.offer(closedNotifications, undefined) : Effect.void, + ); + const unload = harness.runtime.unloadThread!({ + providerThread: harness.providerThread, + awaitClosed: true, + }); + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + const closing = yield* unload.pipe(Effect.forkScoped); + yield* Deferred.await(closeReached); + yield* TestClock.adjust("1 second"); + assert.isUndefined(closing.pollUnsafe(), "a pending close must still be waited for"); + yield* Deferred.succeed(releaseClose, undefined); + yield* Fiber.join(closing); + yield* Queue.take(closedNotifications); + + const unclosed = yield* unload.pipe(Effect.forkScoped); + yield* Queue.take(closedNotifications); + yield* TestClock.adjust("1 second"); + assert.isUndefined(unclosed.pollUnsafe(), "the unload must still wait for thread/closed"); + yield* TestClock.adjust("1 second"); + yield* Fiber.join(unclosed); + + yield* unload; + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + // regression: #33747 Codex refuses thread/resume until an unsubscribed thread finishes closing it.effect.each([ { name: "waits for thread/closed after a closing error", closedFirst: false }, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index efa5d0a9c28f1..b2ad2bfa1f316 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1000,6 +1000,8 @@ function describeCodexGoal(goal: OrchestrationV2ProviderGoal): string { // Codex starts the next goal turn milliseconds after the last one completes. const CODEX_GOAL_CONTINUATION_GRACE = "5 seconds"; +// With thread_unload_delay_secs=0 Codex closes an unsubscribed thread within milliseconds. +const CODEX_THREAD_CLOSED_TIMEOUT = "2 seconds"; const providerTurnsForThread = ( providerTurns: ReadonlyArray, @@ -2055,6 +2057,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); /** Completed by thread/closed after this connection unsubscribes from the native thread. */ const threadClosings = new Map>(); + // A launch-arg override or an older Codex keeps an unsubscribed thread loaded; the + // first unload that waits in vain is the last one on this app-server to wait. + let closesOnUnsubscribe = true; /** Codex acceptForSession covers only the exact command or patch, not future commands. */ const sessionApprovals = new Map< string, @@ -7899,10 +7904,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } // A resume may already wait on the pending entry; replacing it would strand that wait. - if (!threadClosings.has(nativeThreadId)) { - threadClosings.set(nativeThreadId, yield* Deferred.make()); + const pending = threadClosings.get(nativeThreadId); + const closing = pending ?? (yield* Deferred.make()); + threadClosings.set(nativeThreadId, closing); + const { status } = yield* client.request("thread/unsubscribe", { + threadId: nativeThreadId, + }); + // An earlier unload's close may still be pending; Codex then answers notSubscribed. + if ( + !unloadInput.awaitClosed || + !closesOnUnsubscribe || + (status !== "unsubscribed" && pending === undefined) + ) { + return; } - yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); + // Codex replies first and frees the thread's writer lock when it emits thread/closed. + const closed = yield* Deferred.await(closing).pipe( + Effect.timeoutOption(CODEX_THREAD_CLOSED_TIMEOUT), + ); + if (Option.isNone(closed)) closesOnUnsubscribe = false; }, (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 3dff759550c78..d80a292ac74cc 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -4693,7 +4693,10 @@ it.effect.each([ yield* resumeSecondThread; assert.equal((yield* Ref.get(state)).resumeCount, ++resumeCount); - assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]); + assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [ + "native-thread", + "native-thread", + ]); const reloadCases = [ { @@ -4735,6 +4738,7 @@ it.effect.each([ assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [ "native-thread", "native-thread", + "native-thread", ]); yield* manager.detach({ providerSessionId, threadId: secondThreadId }); @@ -5223,6 +5227,69 @@ it.effect.each([ }), ); +// regression: #30630 the old Codex account kept the thread's writer lock after an account switch +it.effect( + "ProviderSessionManagerV2 detach unloads a thread whose row already moved to 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, + }); + const otherInstanceId = ProviderInstanceId.make("codex-other"); + const writeProviderThread = Effect.fn(function* (payload: OrchestrationV2ProviderThread) { + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "provider-thread.updated", + threadId, + driver: CODEX_DRIVER, + occurredAt: now, + payload, + }, + ], + }); + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* writeProviderThread(providerThread); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* runtime.resumeThread({ providerThread, threadId, modelSelection, runtimePolicy }); + + // The switch commits the row under the new instance's session, then detaches the old one. + yield* writeProviderThread({ + ...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 c2f7eb767945d..a7cc0327d8bb7 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -3187,7 +3187,7 @@ export const layerWithOptions = ( return [ Option.none<{ readonly entry: LiveSessionEntry; - readonly idleUnloadFiber: Fiber.Fiber | null; + readonly held: IdleThreadUnload | undefined; }>(), current, ] as const; @@ -3223,13 +3223,13 @@ export const layerWithOptions = ( return [ Option.some({ entry: updatedEntry, - idleUnloadFiber: entry.idleThreadUnloads.get(input.threadId)?.fiber ?? null, + held: entry.idleThreadUnloads.get(input.threadId), }), updated, ] as const; }); if (Option.isSome(detachResult)) { - yield* cancelIdleFiber(detachResult.value.idleUnloadFiber); + yield* cancelIdleFiber(detachResult.value.held?.fiber ?? null); } const detached = Option.map(detachResult, (result) => result.entry); // Plain detaches preserve credentials held by live shared providers across workspace handoffs. @@ -3253,6 +3253,15 @@ export const layerWithOptions = ( } const unloadThread = detached.value.exposedRuntime.unloadThread; if (detached.value.supportsMultipleProviderThreads && unloadThread !== undefined) { + // An instance switch moves the row to the new session before this detach runs, + // so the projection no longer lists the thread this session still holds. + const held = Option.getOrUndefined(detachResult)?.held?.providerThread; + if ( + held !== undefined && + !detachedProviderThreads.some((thread) => thread.id === held.id) + ) { + detachedProviderThreads = [...detachedProviderThreads, held]; + } // 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) { @@ -3271,7 +3280,8 @@ export const layerWithOptions = ( ) { return; } - yield* unloadThread({ providerThread }).pipe( + // The session taking over may run in another provider process. + yield* unloadThread({ providerThread, awaitClosed: true }).pipe( Effect.timeout(UNLOAD_THREAD_TIMEOUT_MS), Effect.catchCause((cause) => Effect.logWarning("orchestration-v2.driver-session.detach-unload-failed", { diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 5b1b6301fa48b..6854576bd570e 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -742,6 +742,8 @@ export interface ProviderAdapterV2SessionRuntime { readonly providerThread: OrchestrationV2ProviderThread; /** Also hide it from the provider's own clients where it can; resuming it must restore it. */ readonly archive?: boolean; + /** Return once the provider has let go of the thread, so another process can resume it. */ + readonly awaitClosed?: boolean; }) => Effect.Effect; readonly respondToRuntimeRequest: ( input: ProviderAdapterV2RuntimeRequestResponseInput,