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

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";

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<ProviderRequestKind> }
         >();
+        const threadCloses = new Map<string, Set<Deferred.Deferred<void>>>();
         /** 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<void>();
+              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(

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

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<void, never> | 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;

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

apps/server/src/provider/codexLaunchArgs.ts

@@ -9,10 +9,17 @@ export const resolveCodexLaunchArgs = (
 
 const codexLaunchArgv = (launchArgs?: string): ReadonlyArray<string> => 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);