diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 884e953a495d6..76c10382b3cb8 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1146,6 +1146,7 @@ describe("CodexAdapterV2 session initialize", () => { runtimePolicy: CODEX_TEST_RUNTIME_POLICY, }), initializeRequests: () => initializeRequests, + events: runtime.events, }; }); @@ -1293,6 +1294,89 @@ 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, + NodeCrypto.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); + 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)); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.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 7de690c5e47d0..f7d3b49a687e5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -245,6 +245,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 CodexThreadSampleResponse = Schema.Struct({ text: Schema.NullOr(Schema.String) }); const decodeCodexThreadSampleResponse = Schema.decodeUnknownEffect(CodexThreadSampleResponse); export const CODEX_PROMPT_SUGGESTION_PROMPT = @@ -1929,7 +1931,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ), ); + const events = yield* Queue.unbounded< + ProviderAdapter.ProviderAdapterV2Event, + 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( @@ -1938,6 +1948,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", { @@ -1953,6 +1965,23 @@ 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 Ref.set(initializeTimeout, error).pipe( + Effect.andThen(Queue.fail(events, error)), + Effect.andThen(Effect.fail(error)), + ); + }, + }), ); const version = userAgent && /\/(\S+)/u.exec(userAgent)?.[1]; const advisory = @@ -1982,10 +2011,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(