From 31efc00e8b6cebcc28f2d41003ffae30000a6dbe Mon Sep 17 00:00:00 2001 From: Stephen Jason Wang Date: Tue, 8 Sep 2026 23:12:19 +0800 Subject: [PATCH] fix(server): drop Codex sessions whose runtime exited The Codex adapter forwarded a runtime's session.exited event but left the dead session in its map, so hasSession and listSessions kept reporting the thread as live. Startup reconciliation trusts listSessions and the session reaper skips sessions holding an activeTurnId, so a thread whose Codex process died mid-turn stayed running and showed Working forever. --- .../src/provider/Layers/CodexAdapter.test.ts | 66 ++++++++++++++++++- .../src/provider/Layers/CodexAdapter.ts | 43 +++++++----- 2 files changed, 92 insertions(+), 17 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index f7c6036885d9..0d7840e18cd4 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -75,6 +75,7 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { updatedAt: this.now, } satisfies ProviderSession), ); + public startEffect: Effect.Effect | undefined; public readonly sendTurnImpl = vi.fn( (_input: CodexSessionRuntimeSendTurnInput): Promise => @@ -127,7 +128,7 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { } start() { - return Effect.promise(() => this.startImpl()); + return this.startEffect ?? Effect.promise(() => this.startImpl()); } getSession = Effect.promise(() => this.startImpl()); @@ -188,6 +189,7 @@ function makeRuntimeFactory() { function makeScopedRuntimeFactory(options?: { readonly failConstruction?: boolean }) { const runtimes: Array = []; const releasedThreadIds: Array = []; + let exitDuringNextStart = false; const factory = vi.fn((runtimeOptions: CodexSessionRuntimeOptions) => Effect.gen(function* () { @@ -206,6 +208,24 @@ function makeScopedRuntimeFactory(options?: { readonly failConstruction?: boolea } const runtime = new FakeCodexRuntime(runtimeOptions); + if (exitDuringNextStart) { + exitDuringNextStart = false; + const startImpl = runtime.startImpl.getMockImplementation(); + runtime.startEffect = runtime + .emit({ + id: asEventId("evt-session-exited-during-start"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "session/exited", + threadId: runtimeOptions.threadId, + message: "Codex App Server exited with code 1.", + }) + .pipe( + Effect.andThen(Effect.yieldNow), + Effect.andThen(Effect.promise(() => startImpl!())), + ); + } runtimes.push(runtime); return runtime; }), @@ -214,6 +234,9 @@ function makeScopedRuntimeFactory(options?: { readonly failConstruction?: boolea return { factory, releasedThreadIds, + exitDuringNextStart: () => { + exitDuringNextStart = true; + }, get lastRuntime(): FakeCodexRuntime | undefined { return runtimes.at(-1); }, @@ -2562,6 +2585,47 @@ const scopedLifecycleLayer = it.layer( ); scopedLifecycleLayer("CodexAdapterLive scoped lifecycle", (it) => { + it.effect("cleans up an exit during startup without deleting its replacement", () => + Effect.gen(function* () { + scopedLifecycleRuntimeFactory.releasedThreadIds.length = 0; + scopedLifecycleRuntimeFactory.exitDuringNextStart(); + const adapter = yield* CodexAdapter; + const threadId = asThreadId("thread-exited"); + const exitedFiber = yield* Stream.runHead( + Stream.filter(adapter.streamEvents, (event) => event.threadId === threadId), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + }); + + const runtime = scopedLifecycleRuntimeFactory.lastRuntime; + NodeAssert.ok(runtime); + const exited = yield* Fiber.join(exitedFiber); + NodeAssert.equal(exited._tag, "Some"); + NodeAssert.equal(yield* adapter.hasSession(threadId), false); + NodeAssert.deepStrictEqual(yield* adapter.listSessions(), []); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + }); + const replacement = scopedLifecycleRuntimeFactory.lastRuntime; + NodeAssert.ok(replacement); + NodeAssert.notEqual(replacement, runtime); + + // Teardown is forked off the event consumer, so let that fiber run. + yield* Effect.yieldNow; + NodeAssert.equal(yield* adapter.hasSession(threadId), true); + NodeAssert.equal((yield* adapter.listSessions()).length, 1); + NodeAssert.equal(runtime.closeImpl.mock.calls.length, 1); + NodeAssert.deepStrictEqual(scopedLifecycleRuntimeFactory.releasedThreadIds, [threadId]); + }), + ); + it.effect("closes the externally owned session scope on stopSession", () => Effect.gen(function* () { scopedLifecycleRuntimeFactory.releasedThreadIds.length = 0; diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 2d88e58dc1fb..d65b806b73e7 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -2231,6 +2231,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( : undefined); const managedNativeEventLogger = options?.nativeEventLogger === undefined ? nativeEventLogger : undefined; + const adapterScope = yield* Scope.Scope; const runtimeEventQueue = yield* Queue.unbounded(); const sessions = new Map(); @@ -2436,9 +2437,31 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( return; } yield* Queue.offerAll(runtimeEventQueue, runtimeEvents); + + // A runtime that reported its own exit is gone, but startup + // reconciliation and the session reaper keep reading the thread as + // live until it leaves this map. Teardown is forked because it + // interrupts this fiber and closes the scope it runs in. + if (runtimeEvents.some((runtimeEvent) => runtimeEvent.type === "session.exited")) { + const exited = sessions.get(input.threadId); + if (exited?.scope === sessionScope) { + sessions.delete(input.threadId); + yield* stopSessionInternal(exited).pipe(Effect.forkIn(adapterScope)); + } + } }), ).pipe(Effect.forkIn(sessionScope)); + const session: CodexAdapterSessionContext = { + threadId: input.threadId, + scope: sessionScope, + runtime, + eventFiber, + turnTokenUsage, + stopped: false, + }; + sessions.set(input.threadId, session); + const started = yield* runtime.start().pipe( Effect.mapError( (cause) => @@ -2449,23 +2472,9 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( cause, }), ), - Effect.onError(() => - runtime.close.pipe( - Effect.andThen(Effect.ignore(Scope.close(sessionScope, Exit.void))), - Effect.andThen(Fiber.interrupt(eventFiber)), - Effect.ignore, - ), - ), + Effect.onError(() => stopSessionInternal(session)), ); - sessions.set(input.threadId, { - threadId: input.threadId, - scope: sessionScope, - runtime, - eventFiber, - turnTokenUsage, - stopped: false, - }); sessionScopeTransferred = true; return started; @@ -2667,7 +2676,9 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( return; } session.stopped = true; - sessions.delete(session.threadId); + if (sessions.get(session.threadId) === session) { + sessions.delete(session.threadId); + } yield* session.runtime.close.pipe(Effect.ignore); yield* Effect.ignore(Scope.close(session.scope, Exit.void)); yield* Fiber.interrupt(session.eventFiber).pipe(Effect.ignore);