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