From d07a77acd8289de88eb55d8557ca6e388dbae4a1 Mon Sep 17 00:00:00 2001 From: TimeAndLocation <112066335+TimeAndLocation@users.noreply.github.com> Date: Tue, 13 Oct 2026 04:29:18 -0700 Subject: [PATCH] fix(codex): terminate event streams when the app-server exits --- .../Adapters/CodexAdapterV2.test.ts | 23 +++++++ .../Adapters/CodexAdapterV2.ts | 20 +++++- .../src/protocol.test.ts | 65 +++++++++++++++++++ .../effect-codex-app-server/src/protocol.ts | 24 +++++-- 4 files changed, 125 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 14bb6a03d811..903adf83cfc3 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/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" }); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 5a51136396ad..72713a6faad7 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/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(); + 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(undefined); const limitedTurnItems = yield* Ref.make( new Map>(), diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 7249afff1071..cf35fa6eaf01 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/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(); + 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 = []; + 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); + }), + ); }); diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 4a32973a988b..db410851328c 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -50,8 +50,14 @@ export interface CodexAppServerPatchedProtocolOptions { } export interface CodexAppServerPatchedProtocol { - readonly incomingNotifications: Stream.Stream; - readonly incomingRequests: Stream.Stream; + 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>(); - const incomingNotifications = - yield* Queue.sliding(MAX_BUFFERED_RAW_MESSAGES); - const incomingRequests = - yield* Queue.sliding(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()); const nextRequestId = yield* Ref.make(1); const remainder: Array = []; @@ -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(