# Files changed · PR \#22791 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/22791/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(server\): Codex recovers after its app-server process is killed](https://git.cafe/pingdotgg/t3code/pull/22791.md) Version: 24 Head commit: 1afd783c5e4ee4764ccc24bbe1649d25be539cee Base commit: ed3f965e0768f9a77729a17c5145a90ff3ab6210 Comparison base: ed3f965e0768f9a77729a17c5145a90ff3ab6210 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts Status: modified; +10; −1 ```diff @@ -88,8 +88,17 @@ const fs = yield* FileSystem.FileSystem; const text = yield* fs.readFileString( decodeURIComponent(delegatedTaskStatusTranscriptFile.pathname), + ); + const transcript = yield* decodeProviderReplayNdjson( + text.replaceAll(queuedFollowupResult, followupResult), ); + // The recording ends when the recorder closes the app-server while the parent + // turn is still open. This test keeps acting as that parent afterwards, and + // replaying the exit would kill the shared process and fail the parent run. + return { + ...transcript, + entries: transcript.entries.filter((entry) => entry.type !== "runtime_exit"), - return yield* decodeProviderReplayNdjson(text.replaceAll(queuedFollowupResult, followupResult)); + }; }, Effect.provide(NodeServices.layer)); function makeProviderSnapshot(input: { ``` ## apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts Status: modified; +160; −0 ```diff @@ -31,6 +31,7 @@ import { assert, describe, it } from "@effect/vitest"; import * as CodexClient from "effect-codex-app-server/client"; import * as CodexReplay from "effect-codex-app-server/replay"; +import * as CodexError from "effect-codex-app-server/errors"; import type * as CodexSchema from "effect-codex-app-server/schema"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -47,6 +48,11 @@ import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as Sink from "effect/Sink"; +import * as Stdio from "effect/Stdio"; +import * as Exit from "effect/Exit"; +import * as Cause from "effect/Cause"; +import * as PlatformError from "effect/PlatformError"; import { TestClock } from "effect/testing"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; @@ -575,6 +581,7 @@ thread: { id: input.nativeThreadId, sessionId: input.nativeThreadId, + projectId: null, forkedFromId: null, preview: "", ephemeral: false, @@ -667,6 +674,159 @@ : entry; } +describe("CodexAdapterV2 process exit", () => { + it.effect.each([ + { exit: "numeric", active: false }, + { exit: "SIGKILL", active: false }, + { exit: "numeric", active: true }, + { exit: "SIGKILL", active: true }, + ] as const)("retires the shared session after $exit (active turn: $active)", ({ exit, active }) => + Effect.gen(function* () { + const scenario = `process-exit-${exit}-${active}`; + const nativeThreadId = `native-${scenario}`; + const nativeTurnId = `turn-${scenario}`; + const input = yield* Queue.unbounded(); + const emit = (frame: unknown) => + Queue.offer(input, new TextEncoder().encode(JSON.stringify(frame) + "\n")); + const outputEnded = yield* Deferred.make(); + const turnStarted = yield* Deferred.make(); + const turnCompleted = yield* Deferred.make(); + const events: Array = []; + const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "work" }); + const terminationError = + exit === "numeric" + ? new CodexError.CodexAppServerProcessExitedError({ code: 17, pid: 52 }) + : new CodexError.CodexAppServerTransportError({ + operation: "read-process-exit-status", + pid: 52, + cause: PlatformError.systemError({ + _tag: "Unknown", + module: "ChildProcess", + method: "exitCode", + cause: new Error("Process terminated by SIGKILL"), + }), + }); + const client = yield* CodexClient.make( + Stdio.make({ + args: Effect.succeed([]), + stdin: Stream.fromQueue(input), + stdout: () => + Sink.forEach((chunk: string | Uint8Array) => + Effect.gen(function* () { + const frame = JSON.parse( + typeof chunk === "string" ? chunk : new TextDecoder().decode(chunk), + ) as { readonly id?: number; readonly method: string }; + const response = preamble.find( + (entry) => + entry.type === "emit_inbound" && + entry.label === frame.method && + Predicate.isObject(entry.frame) && + "id" in entry.frame, + ); + if (response?.type === "emit_inbound") { + yield* emit(response.frame); + } + }), + ).pipe(Sink.ensuring(Deferred.succeed(outputEnded, undefined))), + stderr: () => Sink.drain, + }), + {}, + Effect.succeed(terminationError), + ); + const adapter = yield* CodexAdapterV2.makeCodexAdapterV2({ + instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID, + settings: DEFAULT_CODEX_SETTINGS, + environment: {}, + clientFactory: { open: () => Effect.succeed(client) }, + crypto: yield* Crypto.Crypto, + fileSystem: yield* FileSystem.FileSystem, + idAllocator: yield* IdAllocator.IdAllocatorV2, + serverConfig: yield* makeReplayServerConfig(scenario).pipe(Effect.orDie), + }); + const threadId = ThreadId.make(`thread-${scenario}`); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make(`session-${scenario}`), + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + const eventReader = yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.sync(() => events.push(event)).pipe( + Effect.andThen( + event.type === "provider_turn.updated" && event.providerTurn.status === "running" + ? Deferred.succeed(turnStarted, undefined) + : event.type === "turn.terminal" + ? Deferred.succeed(turnCompleted, undefined) + : Effect.void, + ), + ), + ), + Effect.forkScoped, + ); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + yield* runtime.startTurn( + makeCodexTestTurnInput({ + threadId, + providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(`attempt-${scenario}`), + text: "work", + }), + ); + yield* emit({ + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), + }, + }); + yield* Deferred.await(turnStarted); + if (!active) { + yield* emit({ + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }); + yield* Deferred.await(turnCompleted); + assert.isTrue( + events.some((event) => event.type === "turn.terminal" && event.status === "completed"), + ); + } + + yield* Queue.end(input); + // Output closes only after the protocol records the process termination. + yield* Deferred.await(outputEnded); + const rpcError = yield* client.request("thread/list", {}).pipe(Effect.flip); + assert.strictEqual(rpcError, terminationError); + // Drain runnable lifecycle work before inspecting the reader; no wall-clock deadline. + yield* TestClock.adjust(0); + const readerExit = eventReader.pollUnsafe(); + // A clean end of stream reads as manual_shutdown to the session manager, so + // only a stream failure retires the dead process as a runtime error. + assert.isTrue( + readerExit !== undefined && + Exit.isFailure(readerExit) && + Schema.is(ProviderAdapter.ProviderAdapterEventStreamError)( + Cause.squash(readerExit.cause), + ), + "the session manager must receive process death so it can retire the shared process", + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); +}); + describe("CodexAdapterV2 session initialize", () => { const openReplaySession = ( transcript: CodexReplay.CodexAppServerReplayTranscript, ``` ## apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts Status: modified; +19; −1 ```diff @@ -1838,7 +1838,25 @@ model: input.modelSelection.model, now, }); + const events = yield* Queue.unbounded< + ProviderAdapter.ProviderAdapterV2Event, + ProviderAdapter.ProviderAdapterEventStreamError - const events = yield* Queue.unbounded(); + >(); + // Drain final notifications before the session manager retires the dead process. + // Queue.fail lets readers take items already buffered before they see the failure. + yield* client.awaitTermination.pipe( + 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>(), ``` ## packages/effect-codex-app-server/src/client.ts Status: modified; +5; −0 ```diff @@ -1,4 +1,5 @@ import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; @@ -40,6 +41,7 @@ CodexAppServerClient, { readonly raw: CodexAppServerClientRaw; + readonly awaitTermination: Effect.Effect; readonly request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], @@ -100,6 +102,7 @@ terminationError?: Effect.Effect, ): Effect.fn.Return { const requestHandlers = new Map(); + const termination = yield* Deferred.make(); const notificationHandlers = new Map>(); let unknownRequestHandler: | ((method: string, params: unknown) => Effect.Effect) @@ -203,6 +206,7 @@ ...(options.logger ? { logger: options.logger } : {}), onNotification: dispatchNotification, onRequest: dispatchRequest, + onTermination: (error) => Deferred.fail(termination, error).pipe(Effect.asVoid), }); const request = ( @@ -230,6 +234,7 @@ ); return CodexAppServerClient.of({ + awaitTermination: Deferred.await(termination), raw: { notifications: transport.incomingNotifications, requests: transport.incomingRequests, ```