From c65c7a6a1afa675fa446f011768524b22a540e36 Mon Sep 17 00:00:00 2001 From: Mahir Abbas Date: Wed, 28 Oct 2026 16:28:08 +0000 Subject: [PATCH 1/2] fix(server): a hung Codex app-server no longer blocks every later Codex run All Codex threads on one instance share one app-server session. If that app-server hangs before it answers `initialize`, nothing times out. During open, the hang holds the session open lock. After open (managed Codex), it holds the initialize permit. Either way every later Codex run waits behind it and stays in "starting" until T3 restarts. `initialize` now times out after 30 seconds and fails the session's event stream. A failed open is discarded, and a failed stream makes the session manager retire the session. Both stop the hung app-server, so the next run spawns a fresh one. The event queue is now created before `initialize` can run during open. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 68 +++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 24 +++++-- 2 files changed, 88 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 2d14fa9518de6..a712dbd419a50 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -945,6 +945,7 @@ describe("CodexAdapterV2 session initialize", () => { runtimePolicy: CODEX_TEST_RUNTIME_POLICY, }), initializeRequests: () => initializeRequests, + events: runtime.events, }; }); @@ -1077,6 +1078,73 @@ describe("CodexAdapterV2 session initialize", () => { ), ); + it.effect("fails the open when initialize never answers during open", () => + Effect.gen(function* () { + const initializeAwaitingResponse = yield* Deferred.make(); + const opened = yield* openReplaySession( + makeCodexReplayTranscript({ + scenario: "initialize-hung-open", + entries: replayPreamble("initialize-hung-open").slice(0, 2), + }), + (entry) => + entry.label === "initialize" + ? Deferred.succeed(initializeAwaitingResponse, undefined).pipe( + Effect.andThen(Effect.never), + ) + : Effect.void, + () => Effect.die("initialize never answers"), + ).pipe(Effect.flip, Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(initializeAwaitingResponse); + yield* TestClock.adjust("30 seconds"); + + // The session manager discards a session whose open fails. + assert.instanceOf(yield* Fiber.join(opened), ProviderAdapter.ProviderAdapterOpenSessionError); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect("retires the shared session when initialize never answers", () => + Effect.gen(function* () { + const initializeAwaitingResponse = yield* Deferred.make(); + const session = yield* openReplaySession( + makeCodexReplayTranscript({ + scenario: "initialize-hung", + entries: replayPreamble("initialize-hung").slice(0, 2), + }), + (entry) => + entry.label === "initialize" + ? Deferred.succeed(initializeAwaitingResponse, undefined).pipe( + Effect.andThen(Effect.never), + ) + : Effect.void, + ); + const eventsExit = yield* Stream.runDrain(session.events).pipe( + Effect.exit, + Effect.forkChild({ startImmediately: true }), + ); + + const ensured = yield* session + .ensureThread("thread-initialize-hung") + .pipe(Effect.flip, Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(initializeAwaitingResponse); + yield* TestClock.adjust("30 seconds"); + + assert.equal((yield* Fiber.join(ensured))._tag, "ProviderAdapterEnsureThreadError"); + // The session manager retires a session whose event stream fails. + const exit = yield* Fiber.join(eventsExit); + assert.isTrue(Exit.isFailure(exit)); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("completes the handshake after a caller is interrupted mid-initialize", () => Effect.gen(function* () { const initializeAwaitingResponse = yield* Deferred.make(); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index f81a3809b35fe..feb606e90c461 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -230,6 +230,8 @@ const decodeCodexBackgroundTerminalTerminateResponse = Schema.decodeUnknownEffec const decodeCodexBackgroundTerminalsListResponse = Schema.decodeUnknownEffect( CodexBackgroundTerminalsListResponse, ); +// Initialize normally answers in under a second; leave room for a slow first launch. +const CODEX_INITIALIZE_TIMEOUT = "30 seconds"; const CODEX_CLIENT_CAPABILITIES = { experimentalApi: true, optOutNotificationMethods: ["turn/diff/updated"], @@ -1733,6 +1735,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ), ); + const events = yield* Queue.unbounded< + ProviderAdapter.ProviderAdapterV2Event, + ProviderAdapter.ProviderAdapterEventStreamError + >(); const initialized = yield* Ref.make(false); // Codex rejects a second initialize on the shared app-server. const initializePermit = yield* Semaphore.make(1); @@ -1757,6 +1763,20 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? Effect.succeed(null) : Effect.fail(error), }), + // A hung app-server never answers, and every thread on this shared + // session waits behind the permit. Failing the event stream retires + // the session, so the next turn spawns a fresh app-server. + Effect.timeoutOrElse({ + duration: CODEX_INITIALIZE_TIMEOUT, + orElse: () => { + const error = new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: CODEX_PROVIDER, + providerSessionId: input.providerSessionId, + cause: `Codex app-server did not answer initialize within ${CODEX_INITIALIZE_TIMEOUT}.`, + }); + return Queue.fail(events, error).pipe(Effect.andThen(Effect.fail(error))); + }, + }), ); const version = userAgent && /\/(\S+)/u.exec(userAgent)?.[1]; const advisory = @@ -1786,10 +1806,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( model: input.modelSelection.model, now, }); - const events = yield* Queue.unbounded< - ProviderAdapter.ProviderAdapterV2Event, - ProviderAdapter.ProviderAdapterEventStreamError - >(); // Drain final notifications before the session manager retires the dead process. // Queue.fail lets readers take items already buffered before they see the failure. yield* client.awaitTermination.pipe( From bf5ad4be141b9e0c0719b2551c12af4a297fb101 Mon Sep 17 00:00:00 2001 From: Mahir Abbas Date: Tue, 3 Nov 2026 09:41:39 +0000 Subject: [PATCH 2/2] fix(server): threads waiting on a timed-out Codex initialize fail at once After the first `initialize` timeout the session is retiring, but a thread waiting on the initialize permit could still send another `initialize` to the same hung app-server and wait up to 30 seconds more. The adapter now records the timeout, and later waiters fail with the same error without a new request. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/Adapters/CodexAdapterV2.test.ts | 6 ++++++ .../src/orchestration-v2/Adapters/CodexAdapterV2.ts | 11 ++++++++++- 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index a712dbd419a50..97547a373a0b5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1131,9 +1131,15 @@ describe("CodexAdapterV2 session initialize", () => { .ensureThread("thread-initialize-hung") .pipe(Effect.flip, Effect.forkChild({ startImmediately: true })); yield* Deferred.await(initializeAwaitingResponse); + const waiting = yield* session + .ensureThread("thread-initialize-hung-waiting") + .pipe(Effect.flip, Effect.forkChild({ startImmediately: true })); yield* TestClock.adjust("30 seconds"); assert.equal((yield* Fiber.join(ensured))._tag, "ProviderAdapterEnsureThreadError"); + // A waiter fails at once instead of re-sending initialize to the hung app-server. + assert.equal((yield* Fiber.join(waiting))._tag, "ProviderAdapterEnsureThreadError"); + assert.equal(session.initializeRequests(), 1); // The session manager retires a session whose event stream fails. const exit = yield* Fiber.join(eventsExit); assert.isTrue(Exit.isFailure(exit)); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index feb606e90c461..d66e74de76a01 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1740,6 +1740,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ProviderAdapter.ProviderAdapterEventStreamError >(); const initialized = yield* Ref.make(false); + // After a timeout the session is retiring; waiters must not re-send to the hung app-server. + const initializeTimeout = yield* Ref.make< + ProviderAdapter.ProviderAdapterEventStreamError | undefined + >(undefined); // Codex rejects a second initialize on the shared app-server. const initializePermit = yield* Semaphore.make(1); const ensureInitialized = initializePermit.withPermit( @@ -1748,6 +1752,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (alreadyInitialized) { return; } + const timedOut = yield* Ref.get(initializeTimeout); + if (timedOut !== undefined) return yield* timedOut; const userAgent = yield* client .request("initialize", { @@ -1774,7 +1780,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( providerSessionId: input.providerSessionId, cause: `Codex app-server did not answer initialize within ${CODEX_INITIALIZE_TIMEOUT}.`, }); - return Queue.fail(events, error).pipe(Effect.andThen(Effect.fail(error))); + return Ref.set(initializeTimeout, error).pipe( + Effect.andThen(Queue.fail(events, error)), + Effect.andThen(Effect.fail(error)), + ); }, }), );