fix(server): drop Codex sessions whose runtime exited

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<ProviderSession> | undefined;
 
   public readonly sendTurnImpl = vi.fn(
     (_input: CodexSessionRuntimeSendTurnInput): Promise<ProviderTurnStartResult> =>
@@ -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<FakeCodexRuntime> = [];
   const releasedThreadIds: Array<ThreadId> = [];
+  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;

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<ProviderRuntimeEvent>();
   const sessions = new Map<ThreadId, CodexAdapterSessionContext>();
 
@@ -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);