From 03ba09b683f5c19cd262d23aa74e00f98192826b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 19:47:47 +0000 Subject: [PATCH 1/2] fix(server): switching Codex accounts releases the old account's writer Switching a thread to another Codex instance moves its provider-thread row to the new session in the same commit that queues the old session's detach. Detach picked the threads to unload from the projection by session id, so it found none, sent no thread/unsubscribe and cancelled the idle unload. The old account's app-server kept the native thread and its writer lock, and the new account's resume failed with "already has an active writer". Detach now also unloads the thread the session recorded holding for the app thread. A detach unload then waits for Codex's thread/closed, which is when the writer lock is freed, for at most two seconds. An app-server that lets that wait run out (a launch-arg override of thread_unload_delay_secs, or a Codex too old to have it) is not waited on again. Fixes pingdotgg/t3code#30630 Co-Authored-By: Claude Fable 5.2 --- .../Adapters/CodexAdapterV2.test.ts | 77 +++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 23 +++++- .../ProviderSessionManager.test.ts | 69 ++++++++++++++++- .../ProviderSessionManager.ts | 18 ++++- .../src/server/ProviderAdapter.ts | 2 + 5 files changed, 181 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 09ec3034a9bc9..284a4aea7569e 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -3581,6 +3581,83 @@ 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): 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: "unsubscribed" } }, + }, + ]; + const closedFrame = (threadId: string): CodexReplay.CodexAppServerReplayEntry => ({ + type: "emit_inbound", + label: `closed-${threadId}`, + frame: { method: "thread/closed", params: { threadId } }, + }); + const closedNotifications = yield* Queue.unbounded(); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "codex-detach-unload-awaits-closed", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }).slice(0, 5), + ...unsubscribe(3), + closedFrame(nativeThreadId), + ...unsubscribe(4), + // Replies fence notifications on the protocol reader, so this one marks reply 4 as read. + closedFrame("unrelated-thread"), + ...unsubscribe(5), + ], + }), + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + (method) => + method === "thread/closed" ? Queue.offer(closedNotifications, undefined) : Effect.void, + ); + const unload = harness.runtime.unloadThread!({ + providerThread: harness.providerThread, + awaitClosed: true, + }); + yield* unload; + 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..ed02b0d5e1689 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,22 @@ 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()); + let closing = threadClosings.get(nativeThreadId); + if (closing === undefined) { + closing = yield* Deferred.make(); + threadClosings.set(nativeThreadId, closing); } - yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); + const { status } = yield* client.request("thread/unsubscribe", { + threadId: nativeThreadId, + }); + if (!unloadInput.awaitClosed || status !== "unsubscribed" || !closesOnUnsubscribe) { + return; + } + // 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, From aba7a446534dbbd8c11118ebe1c229a50a8c6f1c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:11:05 +0000 Subject: [PATCH 2/2] fix(server): a detach unload waits for a Codex thread close that is already pending An idle unload unsubscribes without waiting for thread/closed. A detach that unloads the same thread before the close arrives gets notSubscribed from Codex and returned at once, while the writer lock could still be held. It now waits whenever a close was already pending. Co-Authored-By: Claude Fable 5.2 --- .../Adapters/CodexAdapterV2.test.ts | 32 +++++++++++++++---- .../Adapters/CodexAdapterV2.ts | 15 +++++---- 2 files changed, 34 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 284a4aea7569e..64b72b1655aa8 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -3587,7 +3587,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { () => Effect.gen(function* () { const nativeThreadId = "handed-over-thread"; - const unsubscribe = (id: number): ReadonlyArray => [ + const unsubscribe = ( + id: number, + status = "unsubscribed", + ): ReadonlyArray => [ { type: "expect_outbound", label: `unsubscribe-${id}`, @@ -3596,7 +3599,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { { type: "emit_inbound", label: `unsubscribe-result-${id}`, - frame: { id, result: { status: "unsubscribed" } }, + frame: { id, result: { status } }, }, ]; const closedFrame = (threadId: string): CodexReplay.CodexAppServerReplayEntry => ({ @@ -3605,6 +3608,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { 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", @@ -3614,18 +3619,25 @@ describe("CodexAdapterV2 post-settle continuation", () => { 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(4), - // Replies fence notifications on the protocol reader, so this one marks reply 4 as read. - closedFrame("unrelated-thread"), ...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, - undefined, + (entry) => + entry.label === `closed-${nativeThreadId}` + ? Deferred.succeed(closeReached, undefined).pipe( + Effect.andThen(Deferred.await(releaseClose)), + ) + : Effect.void, undefined, undefined, (method) => @@ -3635,7 +3647,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { providerThread: harness.providerThread, awaitClosed: true, }); - yield* unload; + 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); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index ed02b0d5e1689..b2ad2bfa1f316 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7904,15 +7904,18 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } // A resume may already wait on the pending entry; replacing it would strand that wait. - let closing = threadClosings.get(nativeThreadId); - if (closing === undefined) { - closing = yield* Deferred.make(); - threadClosings.set(nativeThreadId, closing); - } + const pending = threadClosings.get(nativeThreadId); + const closing = pending ?? (yield* Deferred.make()); + threadClosings.set(nativeThreadId, closing); const { status } = yield* client.request("thread/unsubscribe", { threadId: nativeThreadId, }); - if (!unloadInput.awaitClosed || status !== "unsubscribed" || !closesOnUnsubscribe) { + // An earlier unload's close may still be pending; Codex then answers notSubscribed. + if ( + !unloadInput.awaitClosed || + !closesOnUnsubscribe || + (status !== "unsubscribed" && pending === undefined) + ) { return; } // Codex replies first and frees the thread's writer lock when it emits thread/closed.