fix(codex): terminate event streams when the app-server exits

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -1802,6 +1802,7 @@ describe("CodexAdapterV2 session initialize", () => {
         runtimePolicy: CODEX_TEST_RUNTIME_POLICY,
       });
       return {
+        runtime,
         ensureThread: (threadId: string) =>
           runtime.ensureThread({
             threadId: ThreadId.make(threadId),
@@ -1812,6 +1813,28 @@ describe("CodexAdapterV2 session initialize", () => {
       };
     });
 
+  it.effect("fails the runtime event stream when its app-server exits", () =>
+    Effect.gen(function* () {
+      const session = yield* openReplaySession(
+        makeCodexReplayTranscript({
+          scenario: "session-process-exit",
+          entries: [{ type: "runtime_exit", status: "success" }],
+        }),
+      );
+      const error = yield* session.runtime.events.pipe(Stream.runDrain, Effect.flip);
+      assert.instanceOf(error, ProviderAdapter.ProviderAdapterEventStreamError);
+      assert.equal(error.driver, CodexAdapterV2.CODEX_DRIVER_KIND);
+      assert.equal(error.providerSessionId, "provider-session-session-process-exit");
+      assert.instanceOf(error.cause, CodexError.CodexAppServerProcessExitedError);
+      assert.equal(error.cause.code, 0);
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(
+        Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+      ),
+    ),
+  );
+
   const replayPreamble = (nativeThreadId: string) =>
     codexReplayPreamble({ nativeThreadId, nativeTurnId: "unused", prompt: "unused" });
 

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -1818,7 +1818,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
           model: input.modelSelection.model,
           now,
         });
-        const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
+        const events = yield* Queue.unbounded<
+          ProviderAdapter.ProviderAdapterV2Event,
+          ProviderAdapter.ProviderAdapterEventStreamError
+        >();
+        // Drain translated events before the session manager releases the exited process.
+        yield* client.raw.notifications.pipe(
+          Stream.runDrain,
+          Effect.catch((cause) =>
+            Queue.fail(
+              events,
+              new ProviderAdapter.ProviderAdapterEventStreamError({
+                driver: CODEX_PROVIDER,
+                providerSessionId: input.providerSessionId,
+                cause,
+              }),
+            ),
+          ),
+          Effect.forkScoped,
+        );
         const rateLimitSnapshot = yield* Ref.make<CodexRateLimitSnapshot | undefined>(undefined);
         const limitedTurnItems = yield* Ref.make(
           new Map<ProviderThreadId, Extract<OrchestrationV2TurnItem, { type: "error" }>>(),

packages/effect-codex-app-server/src/protocol.test.ts

@@ -1,5 +1,6 @@
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
 import * as Queue from "effect/Queue";
 import * as Schema from "effect/Schema";
@@ -804,4 +805,68 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
       assert.equal("cause" in error, false);
     }),
   );
+
+  it.effect("fails pending requests and both incoming streams with the process exit", () =>
+    Effect.gen(function* () {
+      const { stdio, input, output } = yield* makeInMemoryStdio();
+      const expected = new CodexError.CodexAppServerProcessExitedError({
+        code: 4294967295,
+        pid: 51,
+      });
+      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
+        stdio,
+        terminationError: Effect.succeed(expected),
+      });
+      const notifications = yield* transport.incomingNotifications.pipe(
+        Stream.runDrain,
+        Effect.flip,
+        Effect.forkScoped,
+      );
+      const requests = yield* transport.incomingRequests.pipe(
+        Stream.runDrain,
+        Effect.flip,
+        Effect.forkScoped,
+      );
+      const pending = yield* transport
+        .request("thread/start", {})
+        .pipe(Effect.exit, Effect.forkScoped);
+      yield* Queue.take(output);
+      yield* Queue.end(input);
+
+      assert.deepEqual(yield* Fiber.join(pending), Exit.fail(expected));
+      assert.strictEqual(yield* Fiber.join(notifications), expected);
+      assert.strictEqual(yield* Fiber.join(requests), expected);
+      assert.deepEqual(
+        yield* transport.request("thread/start", {}).pipe(Effect.exit),
+        Exit.fail(expected),
+      );
+      assert.strictEqual(
+        yield* transport.incomingNotifications.pipe(Stream.runDrain, Effect.flip),
+        expected,
+      );
+    }),
+  );
+
+  it.effect("drains buffered notifications before reporting an input stream ending", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const received = yield* Deferred.make<void>();
+      const notification = { method: "thread/started", params: { thread: { id: "thread-1" } } };
+      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
+        stdio,
+        onNotification: () => Deferred.succeed(received, undefined).pipe(Effect.asVoid),
+      });
+      yield* Queue.offer(input, encodeJsonl(notification));
+      yield* Deferred.await(received);
+      yield* Queue.end(input);
+
+      const buffered: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];
+      const error = yield* transport.incomingNotifications.pipe(
+        Stream.runForEach((value) => Effect.sync(() => buffered.push(value))),
+        Effect.flip,
+      );
+      assert.deepEqual(buffered, [notification]);
+      assert.instanceOf(error, CodexError.CodexAppServerInputStreamEndedError);
+    }),
+  );
 });

packages/effect-codex-app-server/src/protocol.ts

@@ -50,8 +50,14 @@ export interface CodexAppServerPatchedProtocolOptions {
 }
 
 export interface CodexAppServerPatchedProtocol {
-  readonly incomingNotifications: Stream.Stream<CodexAppServerIncomingNotification>;
-  readonly incomingRequests: Stream.Stream<CodexAppServerIncomingRequest>;
+  readonly incomingNotifications: Stream.Stream<
+    CodexAppServerIncomingNotification,
+    CodexError.CodexAppServerError
+  >;
+  readonly incomingRequests: Stream.Stream<
+    CodexAppServerIncomingRequest,
+    CodexError.CodexAppServerError
+  >;
   readonly request: (
     method: string,
     payload?: unknown,
@@ -158,10 +164,14 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
     const protocolScope = yield* Scope.Scope;
     const requestHandlerScope = yield* Scope.fork(protocolScope, "parallel");
     const outgoing = yield* Queue.unbounded<string, Cause.Done<void>>();
-    const incomingNotifications =
-      yield* Queue.sliding<CodexAppServerIncomingNotification>(MAX_BUFFERED_RAW_MESSAGES);
-    const incomingRequests =
-      yield* Queue.sliding<CodexAppServerIncomingRequest>(MAX_BUFFERED_RAW_MESSAGES);
+    const incomingNotifications = yield* Queue.sliding<
+      CodexAppServerIncomingNotification,
+      CodexError.CodexAppServerError
+    >(MAX_BUFFERED_RAW_MESSAGES);
+    const incomingRequests = yield* Queue.sliding<
+      CodexAppServerIncomingRequest,
+      CodexError.CodexAppServerError
+    >(MAX_BUFFERED_RAW_MESSAGES);
     const pending = yield* Ref.make(new Map<string, CodexAppServerPendingRequest>());
     const nextRequestId = yield* Ref.make(1);
     const remainder: Array<string> = [];
@@ -203,6 +213,8 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
             const error = yield* classify();
             yield* Ref.set(terminationFailure, Option.some(error));
             yield* failAllPending(error);
+            yield* Queue.fail(incomingNotifications, error);
+            yield* Queue.fail(incomingRequests, error);
             yield* Queue.end(outgoing);
             yield* Deferred.succeed(terminationSignal, undefined);
             yield* Scope.close(requestHandlerScope, Exit.void).pipe(