diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 10e979c650108..f470d0b25bfae 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -5,6 +5,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stdio from "effect/Stdio"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; @@ -644,4 +645,49 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { assert.notInclude(error.message, secret); }), ); + + it.effect("fails pending requests when input ends while the process is still running", () => + Effect.gen(function* () { + const { stdio, input, output } = yield* makeInMemoryStdio(); + const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + terminationError: Effect.never, + }); + const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped); + yield* Queue.take(output); + + yield* Queue.end(input); + yield* TestClock.adjust("249 millis"); + assert.isUndefined(pending.pollUnsafe()); + yield* TestClock.adjust("1 millis"); + + const failure = yield* Fiber.join(pending).pipe( + Effect.match({ + onFailure: (error) => error, + onSuccess: () => assert.fail("Expected the pending request to fail"), + }), + ); + assert.instanceOf(failure, CodexError.CodexAppServerInputStreamEndedError); + }), + ); + + it.effect("reports a process exit that follows input end", () => + Effect.gen(function* () { + const { stdio, input } = yield* makeInMemoryStdio(); + const exitError = new CodexError.CodexAppServerProcessExitedError({ code: 1, pid: 123 }); + const exited = yield* Deferred.make(); + const termination = yield* Deferred.make(); + yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + terminationError: Deferred.await(exited), + onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid), + }); + + yield* Queue.end(input); + yield* TestClock.adjust("100 millis"); + yield* Deferred.succeed(exited, exitError); + + assert.strictEqual(yield* Deferred.await(termination), exitError); + }), + ); }); diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index bf67174033f4d..b0ff5436e9e77 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -18,6 +18,7 @@ const isCodexAppServerError = Schema.is(CodexError.CodexAppServerError); const MAX_BUFFERED_RAW_MESSAGES = 32; // Bound JSONL lines before joining and parsing; the LF delimiter is excluded. const MAX_INCOMING_MESSAGE_BYTES = 128 * 1024 * 1024; +const EXIT_STATUS_GRACE = "250 millis"; export interface CodexAppServerProtocolLogEvent { readonly direction: "incoming" | "outgoing"; @@ -445,11 +446,20 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Effect.matchEffect({ onFailure: (error) => handleTermination(() => Effect.succeed(error)), onSuccess: () => - handleTermination( - () => - options.terminationError ?? - Effect.succeed(new CodexError.CodexAppServerInputStreamEndedError({})), - ), + handleTermination(() => { + const ended = new CodexError.CodexAppServerInputStreamEndedError({}); + // Node ends stdout before it reports the exit of a crashed process, so + // wait briefly for the exit status. stdout can also close while the + // process keeps running, so the wait must not hang pending requests. + return options.terminationError === undefined + ? Effect.succeed(ended) + : options.terminationError.pipe( + Effect.timeoutOrElse({ + duration: EXIT_STATUS_GRACE, + orElse: () => Effect.succeed(ended), + }), + ); + }), }), ), }),