fix(server): switching Codex accounts releases the old account's writer

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<CodexReplay.CodexAppServerReplayEntry> => [
+          {
+            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<void>();
+        const closeReached = yield* Deferred.make<void>();
+        const releaseClose = yield* Deferred.make<void>();
+        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 },

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<OrchestrationV2ProviderTurn>,
@@ -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<string, Deferred.Deferred<void>>();
+        // 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<void>());
+              const pending = threadClosings.get(nativeThreadId);
+              const closing = pending ?? (yield* Deferred.make<void>());
+              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(

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",
   () =>

apps/server/src/orchestration-v2/ProviderSessionManager.ts

@@ -3187,7 +3187,7 @@ export const layerWithOptions = (
                 return [
                   Option.none<{
                     readonly entry: LiveSessionEntry;
-                    readonly idleUnloadFiber: Fiber.Fiber<void, never> | 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", {

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<void, ProviderAdapterV2Error>;
   readonly respondToRuntimeRequest: (
     input: ProviderAdapterV2RuntimeRequestResponseInput,