Keep concurrent Codex discovery from failing during startup (#9008)

* Serialize Codex app-server startup by state directory

* Keep Codex session constructor unchanged

* Type Codex startup failures and inject process launch port

* fix(codex): retain startup diagnostics emitted after exit

Classify failed initialization after stdio closes so the bounded SQLite retry sees the complete stderr. Keep running-session exit handling immediate. The regression delivers the SQLite diagnostic between exit and close.

* test(codex): inject history test process through the public dependency

The history fixture still replaced the old private spawn method, so provider launch preparation tried to find a real CLI in CI. Inject the same process interface as other provider tests and use an invalid command as a guard against accidental real launches.

---------

Co-authored-by: Michael Wu <michaelmwu@gmail.com>
Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
Co-authored-by: Ivan Malison <1246619+colonelpanic8@users.noreply.github.com>
This commit is contained in:
paseo-bot[bot]
2026-11-07 14:03:47 +00:00
committed by GitHub
co-authored by Michael Wu Ivan Malison
parent 28c01ea1ce
commit a48bc3aa6f
11 changed files with 903 additions and 131 deletions
@@ -3,7 +3,7 @@ import { execFileSync } from "node:child_process";
import os from "node:os";
import path from "node:path";
import { createServer } from "node:http";
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { once } from "node:events";
import { AgentManager } from "../agent-manager.js";
@@ -217,6 +217,33 @@ function waitForEvent<TEvent extends AgentStreamEvent>(params: {
}
describe("Codex app-server provider (local e2e)", () => {
test.runIf(isCodexInstalled())(
"concurrent import scans initialize successfully in a fresh shared Codex home",
async () => {
const artifacts = path.resolve(".artifacts");
mkdirSync(artifacts, { recursive: true });
const root = mkdtempSync(path.join(artifacts, "codex-concurrent-startup-"));
try {
for (let round = 0; round < 4; round++) {
const codexHome = path.join(root, String(round));
mkdirSync(codexHome);
const provider = new CodexAppServerAgentClient(createTestLogger(), {
env: { CODEX_HOME: codexHome },
});
const results = await Promise.allSettled(
Array.from({ length: 24 }, () => provider.listImportableSessions({ limit: 1 })),
);
expect(results).toEqual(
Array.from({ length: 24 }, () => ({ status: "fulfilled", value: [] })),
);
}
} finally {
rmSync(root, { recursive: true, force: true });
}
},
60_000,
);
test.runIf(isCodexInstalled())(
"reloads a persisted idle thread repeatedly without overlapping writers",
async () => {
@@ -587,6 +587,10 @@ test("concurrent session APIs share one reconnect after an idle provider exit",
createFakeCodexAppServer(),
];
let spawnCount = 0;
let markReconnectStarted!: () => void;
const reconnectStarted = new Promise<void>((resolve) => {
markReconnectStarted = resolve;
});
let releaseReconnect: (() => void) | undefined;
const reconnectGate = new Promise<void>((resolve) => {
releaseReconnect = resolve;
@@ -602,6 +606,7 @@ test("concurrent session APIs share one reconnect after an idle provider exit",
}
spawnCount += 1;
if (spawnCount > 1) {
markReconnectStarted();
await reconnectGate;
}
return appServer.child;
@@ -614,12 +619,14 @@ test("concurrent session APIs share one reconnect after an idle provider exit",
const runtimeInfo = session.getRuntimeInfo();
const turnStart = session.startTurn("continue after reconnect");
await reconnectStarted;
const reconnectSpawnCount = spawnCount - 1;
releaseReconnect?.();
await expect(runtimeInfo).resolves.toMatchObject({ provider: "codex" });
await expect(turnStart).resolves.toMatchObject({ turnId: expect.any(String) });
expect(reconnectSpawnCount).toBe(1);
expect(spawnCount).toBe(2);
} finally {
releaseReconnect?.();
await session.close();
@@ -35,15 +35,14 @@ describe("CodexAppServerAgentClient spawn error handling", () => {
});
test("fetchCatalog settles when the refresh deadline aborts an unanswered initialize", async () => {
const client = new CodexAppServerAgentClient(logger, {
command: {
mode: "replace",
argv: ["/nonexistent/codex-binary-that-does-not-exist"],
},
});
const child = createCodexAppServerChildProcess();
const kill = vi.spyOn(child, "kill");
Object.assign(client, { spawnAppServer: async () => child });
const client = new CodexAppServerAgentClient(logger, undefined, {
appServerProcess: {
resolvePrefix: async () => ({ command: "codex", args: [] }),
spawn: () => child,
},
});
await expect(
runProviderRefreshWithDeadline({
@@ -87,6 +87,7 @@ describe("Codex executable discovery", () => {
import { CodexAppServerClient } from "./codex/app-server-transport.js";
import {
createCodexAppServerChildProcess,
createFakeCodexAppServer,
fakeCodexThreadHistory,
type FakeCodexAppServer,
@@ -185,18 +186,24 @@ function createProviderWithFakeAppServer(
): CodexAppServerAgentClient {
const provider = new CodexAppServerAgentClient(createTestLogger(), options?.runtimeSettings, {
customProvider: options?.customProvider,
appServerProcess: createFakeCodexAppServerProcess(() => appServer.child),
});
const internals = castInternals<{
goalsEnabledPromise: Promise<boolean> | null;
autoReviewEnabledPromise: Promise<boolean> | null;
spawnAppServer: () => Promise<ChildProcessWithoutNullStreams>;
}>(provider);
internals.goalsEnabledPromise = Promise.resolve(false);
internals.autoReviewEnabledPromise = Promise.resolve(false);
internals.spawnAppServer = async () => appServer.child;
return provider;
}
function createFakeCodexAppServerProcess(spawn: () => ChildProcessWithoutNullStreams) {
return {
resolvePrefix: async () => ({ command: "codex", args: [] }),
spawn: () => spawn(),
};
}
async function startPublicSteeringSession(
appServer: FakeCodexAppServer,
resolveSlashCommandInvocation?: (prompt: AgentPromptInput) => Promise<{
@@ -626,24 +633,25 @@ function codexAgentItem(text: string) {
// app-server serving `threads` from thread/read.
function createReopenableCodexAgent(threads: Record<string, unknown[]>) {
const appServers: FakeCodexAppServer[] = [];
const provider = new CodexAppServerAgentClient(createTestLogger());
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
appServerProcess: createFakeCodexAppServerProcess(() => {
const appServer = createFakeCodexAppServer({
"thread/read": (params) => {
const { threadId } = params as { threadId: string };
return { thread: { id: threadId, turns: threads[threadId] ?? [] } };
},
});
appServers.push(appServer);
return appServer.child;
}),
});
const internals = castInternals<{
goalsEnabledPromise: Promise<boolean> | null;
autoReviewEnabledPromise: Promise<boolean> | null;
spawnAppServer: () => Promise<ChildProcessWithoutNullStreams>;
}>(provider);
internals.goalsEnabledPromise = Promise.resolve(false);
internals.autoReviewEnabledPromise = Promise.resolve(false);
internals.spawnAppServer = async () => {
const appServer = createFakeCodexAppServer({
"thread/read": (params) => {
const { threadId } = params as { threadId: string };
return { thread: { id: threadId, turns: threads[threadId] ?? [] } };
},
});
appServers.push(appServer);
return appServer.child;
};
// The fake app-server stands in for the Codex CLI, which test machines may not have.
const client: AgentClient = {
provider: "codex",
@@ -2373,10 +2381,9 @@ describe("Codex app-server provider", () => {
return { thread: { id: "native-thread-id" } };
},
});
const provider = new CodexAppServerAgentClient(createTestLogger());
castInternals<{ spawnAppServer: () => Promise<ChildProcessWithoutNullStreams> }>(
provider,
).spawnAppServer = async () => appServer.child;
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
appServerProcess: createFakeCodexAppServerProcess(() => appServer.child),
});
await provider.unarchiveNativeSession({
provider: "codex",
@@ -2398,10 +2405,9 @@ describe("Codex app-server provider", () => {
return { thread: { id: "native-thread-id" } };
},
});
const provider = new CodexAppServerAgentClient(createTestLogger());
castInternals<{ spawnAppServer: () => Promise<ChildProcessWithoutNullStreams> }>(
provider,
).spawnAppServer = async () => appServer.child;
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
appServerProcess: createFakeCodexAppServerProcess(() => appServer.child),
});
await provider.archiveNativeSession({
provider: "codex",
@@ -2423,10 +2429,9 @@ describe("Codex app-server provider", () => {
return { thread: { id: "persisted-thread-id" } };
},
});
const provider = new CodexAppServerAgentClient(createTestLogger());
castInternals<{ spawnAppServer: () => Promise<ChildProcessWithoutNullStreams> }>(
provider,
).spawnAppServer = async () => appServer.child;
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
appServerProcess: createFakeCodexAppServerProcess(() => appServer.child),
});
await provider.unarchiveNativeSession({
provider: "codex",
@@ -2455,10 +2460,9 @@ describe("Codex app-server provider", () => {
return { thread: { id: "active-thread-id", turns: [] } };
},
});
const provider = new CodexAppServerAgentClient(createTestLogger());
castInternals<{ spawnAppServer: () => Promise<ChildProcessWithoutNullStreams> }>(
provider,
).spawnAppServer = async () => appServer.child;
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
appServerProcess: createFakeCodexAppServerProcess(() => appServer.child),
});
await provider.unarchiveNativeSession({
provider: "codex",
@@ -2488,10 +2492,9 @@ describe("Codex app-server provider", () => {
return Promise.reject(new Error("thread not found"));
},
});
const provider = new CodexAppServerAgentClient(createTestLogger());
castInternals<{ spawnAppServer: () => Promise<ChildProcessWithoutNullStreams> }>(
provider,
).spawnAppServer = async () => appServer.child;
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
appServerProcess: createFakeCodexAppServerProcess(() => appServer.child),
});
await expect(
provider.unarchiveNativeSession({
@@ -7530,6 +7533,52 @@ describe("Codex app-server provider", () => {
});
describe("Codex importable sessions", () => {
test("serializes concurrent app-server startups while allowing all scans to complete", async () => {
const firstEntered = deferred<void>();
const releaseFirst = deferred<void>();
let spawned = 0;
let activeInitializations = 0;
let maxActiveInitializations = 0;
let disposed = 0;
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
appServerProcess: createFakeCodexAppServerProcess(() => {
spawned++;
return createCodexAppServerChildProcess();
}),
_createCodexClient: () => ({
request: async (method) => {
if (method === "initialize") {
activeInitializations++;
maxActiveInitializations = Math.max(maxActiveInitializations, activeInitializations);
if (spawned === 1) {
firstEntered.resolve();
await releaseFirst.promise;
}
activeInitializations--;
return {};
}
if (method === "thread/list") return { data: [] };
throw new Error(`Unexpected Codex method: ${method}`);
},
notify: () => {},
dispose: async () => {
disposed++;
},
}),
});
const scans = Array.from({ length: 24 }, () => provider.listImportableSessions({ limit: 1 }));
await firstEntered.promise;
expect(spawned).toBe(1);
releaseFirst.resolve();
const results = await Promise.all(scans);
expect(results).toEqual(Array.from({ length: 24 }, () => []));
expect(spawned).toBe(24);
expect(maxActiveInitializations).toBe(1);
expect(disposed).toBe(24);
});
const CODEX_THREAD_PAGE_CAP = 100;
// Codex answers thread/list with at most 100 rows per response whatever limit
@@ -7792,19 +7841,8 @@ describe("Codex importable sessions", () => {
const provider = new CodexAppServerAgentClient(createTestLogger(), undefined, {
_createCodexClient: () => fakeClient,
appServerProcess: createFakeCodexAppServerProcess(createCodexAppServerChildProcess),
});
castInternals<{ spawnAppServer: () => Promise<ChildProcessWithoutNullStreams> }>(
provider,
).spawnAppServer = async () => {
const child = new EventEmitter() as ChildProcessWithoutNullStreams;
child.exitCode = 0;
child.signalCode = null;
child.stdin = new PassThrough();
child.stdout = new PassThrough();
child.stderr = new PassThrough();
child.kill = vi.fn(() => true) as ChildProcessWithoutNullStreams["kill"];
return child;
};
const sessions = await provider.listImportableSessions({ cwd: "/workspace/project-a" });
@@ -75,7 +75,7 @@ import {
probeExecutable,
} from "../../../executable-resolution/executable-resolution.js";
import { createPathEquivalenceMatcher } from "../../../utils/path.js";
import { spawnProcess } from "../../../utils/spawn.js";
import { spawnProcess, type SpawnProcessOptions } from "../../../utils/spawn.js";
import { extractCodexTerminalSessionId, nonEmptyString } from "./tool-call-mapper-utils.js";
import {
buildCodexFeatures,
@@ -94,6 +94,7 @@ import {
type CodexThreadRollbackResponse,
type CodexAppServerTraceContext,
} from "./codex/app-server-transport.js";
import { codexAppServerStartup } from "./codex/app-server-startup.js";
import { type CodexUserMessageTurnIndex, revertCodexConversation } from "./codex/rewind.js";
import {
type CodexHistoryRow,
@@ -291,14 +292,31 @@ const CODEX_MODES: AgentMode[] = [
const DEFAULT_CODEX_MODE_ID = "auto";
interface CodexAppServerClientLike {
request(method: string, params?: unknown): Promise<unknown>;
request(method: string, params?: unknown, timeoutMs?: number): Promise<unknown>;
forkThread?(params: CodexThreadForkParams): Promise<CodexThreadForkResponse>;
rollbackThread?(params: CodexThreadRollbackParams): Promise<CodexThreadRollbackResponse>;
notify(method: string, params?: unknown): void;
dispose(): Promise<void>;
}
interface CodexAppServerLaunchPrefix {
command: string;
args: string[];
}
interface CodexAppServerProcessSpawnInput extends CodexAppServerLaunchPrefix {
options: SpawnProcessOptions;
}
interface CodexAppServerProcessPort {
resolvePrefix(options: {
runtimeSettings?: ProviderRuntimeSettings;
}): Promise<CodexAppServerLaunchPrefix>;
spawn(input: CodexAppServerProcessSpawnInput): ChildProcess;
}
interface CodexAppServerAgentDeps {
appServerProcess?: CodexAppServerProcessPort;
environment?: Record<string, string>;
workspaceGitService?: Pick<WorkspaceGitService, "resolveRepoRoot">;
customProvider?: {
@@ -610,10 +628,9 @@ export async function findDefaultCodexBinary(): Promise<string | null> {
return await findCodexMicrosoftStoreBinary();
}
async function resolveCodexLaunchPrefix(runtimeSettings?: ProviderRuntimeSettings): Promise<{
command: string;
args: string[];
}> {
async function resolveCodexLaunchPrefix(
runtimeSettings?: ProviderRuntimeSettings,
): Promise<CodexAppServerLaunchPrefix> {
const launch = await resolveCodexLaunch(runtimeSettings);
const availability = await checkCodexLaunchAvailable(launch);
if (!availability.available) {
@@ -628,6 +645,11 @@ async function resolveCodexLaunchPrefix(runtimeSettings?: ProviderRuntimeSetting
};
}
const defaultCodexAppServerProcess: CodexAppServerProcessPort = {
resolvePrefix: ({ runtimeSettings }) => resolveCodexLaunchPrefix(runtimeSettings),
spawn: ({ command, args, options }) => spawnProcess(command, args, options),
};
async function resolveCodexLaunch(
runtimeSettings?: ProviderRuntimeSettings,
): Promise<ResolvedProviderLaunch> {
@@ -3369,6 +3391,14 @@ export function buildCodexAppServerEnv(
});
}
function codexStartupStateDirKey(environment: NodeJS.ProcessEnv): string {
const home =
process.platform === "win32"
? environment.USERPROFILE || os.homedir()
: environment.HOME || os.homedir();
return path.resolve(environment.CODEX_HOME || path.join(home, ".codex"));
}
function buildCodexAppServerInitializeParams(): {
clientInfo: { name: string; title: string; version: string };
capabilities: { experimentalApi: true; mcpServerOpenaiFormElicitation: true };
@@ -3566,6 +3596,7 @@ export class CodexAppServerAgentSession implements AgentSession {
private unpairedCompactionItemCompletions = 0;
private connectionState: "disconnected" | "history-ready" | "connected" = "disconnected";
private connectionPromise: Promise<void> | null = null;
private readonly startupAbort = new AbortController();
private closed = false;
private collaborationModes: Array<{
name: string;
@@ -3683,8 +3714,15 @@ export class CodexAppServerAgentSession implements AgentSession {
this.connectionState = "history-ready";
return;
}
const child = await this.spawnAppServer();
const client = new CodexAppServerClient(child, this.logger, () => this.traceContext());
const { client, initialized } = await codexAppServerStartup.start({
stateDirKey: codexStartupStateDirKey(this.harnessEnvironment),
spawn: this.spawnAppServer,
createClient: (child) =>
new CodexAppServerClient(child, this.logger, () => this.traceContext()),
initializeParams: buildCodexAppServerInitializeParams(),
logger: this.logger,
signal: this.startupAbort.signal,
});
if (this.closed) {
await client.dispose();
throw this.createClosedError();
@@ -3697,11 +3735,9 @@ export class CodexAppServerAgentSession implements AgentSession {
this.registerRequestHandlers();
try {
const initialized = toObjectRecord(
await client.request("initialize", buildCodexAppServerInitializeParams()),
this.threadRollbackAvailable = codexServerHasThreadRollback(
toObjectRecord(initialized)?.userAgent,
);
this.threadRollbackAvailable = codexServerHasThreadRollback(initialized?.userAgent);
client.notify("initialized", {});
this.speedModels =
CodexModelListResponseSchema.parse(await client.request("model/list", {})).data ?? [];
@@ -4029,11 +4065,16 @@ export class CodexAppServerAgentSession implements AgentSession {
}
private async readArchivedHistory(): Promise<CodexThreadHistory> {
const child = await this.spawnAppServer();
const client = new CodexAppServerClient(child, this.logger, () => this.traceContext());
const { client } = await codexAppServerStartup.start({
stateDirKey: codexStartupStateDirKey(this.harnessEnvironment),
spawn: this.spawnAppServer,
createClient: (child) =>
new CodexAppServerClient(child, this.logger, () => this.traceContext()),
initializeParams: buildCodexAppServerInitializeParams(),
logger: this.logger,
signal: this.startupAbort.signal,
});
try {
await client.request("initialize", buildCodexAppServerInitializeParams());
client.notify("initialized", {});
return await this.readThreadHistory(client);
} finally {
await client.dispose();
@@ -5296,6 +5337,7 @@ export class CodexAppServerAgentSession implements AgentSession {
async close(): Promise<void> {
this.closed = true;
this.startupAbort.abort(this.createClosedError());
this.background.stop();
this.clearPendingPermissions();
this.pendingToolCallDetails.clear();
@@ -7534,31 +7576,55 @@ export class CodexAppServerAgentClient implements AgentClient {
}
}
private async spawnAppServer(
launchEnv?: Record<string, string>,
options?: { goalsEnabled?: boolean; agentId?: string; environment?: Record<string, string> },
): Promise<ChildProcessWithoutNullStreams> {
const launchPrefix = await resolveCodexLaunchPrefix(this.runtimeSettings);
private async prepareAppServerSpawn(options: {
environment: Record<string, string>;
goalsEnabled?: boolean;
agentId?: string;
}): Promise<() => ChildProcessWithoutNullStreams> {
const processPort = this.deps.appServerProcess ?? defaultCodexAppServerProcess;
const launchPrefix = await processPort.resolvePrefix({ runtimeSettings: this.runtimeSettings });
const args = [...launchPrefix.args, "app-server"];
if (options?.goalsEnabled) {
if (options.goalsEnabled) {
args.push("--enable", "goals");
}
this.logger.trace(
{
agentId: options?.agentId,
provider: CODEX_PROVIDER,
launchPrefix,
goalsEnabled: options?.goalsEnabled === true,
},
"provider.codex.spawn",
);
const child = spawnProcess(launchPrefix.command, args, {
detached: process.platform !== "win32",
stdio: ["pipe", "pipe", "pipe"],
baseEnv: options?.environment ?? buildCodexAppServerEnv(this.runtimeSettings, launchEnv),
return () => {
this.logger.trace(
{
agentId: options.agentId,
provider: CODEX_PROVIDER,
launchPrefix,
goalsEnabled: options.goalsEnabled === true,
},
"provider.codex.spawn",
);
const child = processPort.spawn({
command: launchPrefix.command,
args,
options: {
detached: process.platform !== "win32",
stdio: ["pipe", "pipe", "pipe"],
baseEnv: options.environment,
},
});
assertChildWithPipes(child);
return child;
};
}
private async startAppServer(
options: { signal?: AbortSignal } = {},
): Promise<CodexAppServerClient> {
const environment = buildCodexAppServerEnv(this.runtimeSettings);
const spawn = await this.prepareAppServerSpawn({ environment });
const { client } = await codexAppServerStartup.start({
stateDirKey: codexStartupStateDirKey(environment),
spawn,
createClient: (child) => new CodexAppServerClient(child, this.logger),
initializeParams: buildCodexAppServerInitializeParams(),
logger: this.logger,
signal: options.signal,
});
assertChildWithPipes(child);
return child;
return client;
}
async createSession(
@@ -7577,16 +7643,16 @@ export class CodexAppServerAgentClient implements AgentClient {
const goalsEnabled = await this.resolveGoalsEnabled();
const autoReviewEnabled = await this.resolveAutoReviewEnabled();
const environment = buildCodexAppServerEnv(this.runtimeSettings, launchContext?.env);
const spawn = await this.prepareAppServerSpawn({
environment,
goalsEnabled,
agentId: launchContext?.agentId,
});
const session = new CodexAppServerAgentSession(
sessionConfig,
null,
this.logger,
() =>
this.spawnAppServer(launchContext?.env, {
goalsEnabled,
agentId: launchContext?.agentId,
environment,
}),
async () => spawn(),
{ ...this.sessionDeps(launchContext?.env), environment },
options?.persistSession === false,
goalsEnabled,
@@ -7614,16 +7680,16 @@ export class CodexAppServerAgentClient implements AgentClient {
const goalsEnabled = await this.resolveGoalsEnabled();
const autoReviewEnabled = await this.resolveAutoReviewEnabled();
const environment = buildCodexAppServerEnv(this.runtimeSettings, launchContext?.env);
const spawn = await this.prepareAppServerSpawn({
environment,
goalsEnabled,
agentId: launchContext?.agentId,
});
const session = new CodexAppServerAgentSession(
merged,
handle,
this.logger,
() =>
this.spawnAppServer(launchContext?.env, {
goalsEnabled,
agentId: launchContext?.agentId,
environment,
}),
async () => spawn(),
{ ...this.sessionDeps(launchContext?.env), environment },
false,
goalsEnabled,
@@ -7638,15 +7704,19 @@ export class CodexAppServerAgentClient implements AgentClient {
async listImportableSessions(
options?: ListImportableSessionsOptions,
): Promise<ImportableProviderSession[]> {
const child = await this.spawnAppServer();
const client =
this.deps._createCodexClient?.(child, this.logger, () => ({})) ??
new CodexAppServerClient(child, this.logger);
const environment = buildCodexAppServerEnv(this.runtimeSettings);
const spawn = await this.prepareAppServerSpawn({ environment });
const { client } = await codexAppServerStartup.start({
stateDirKey: codexStartupStateDirKey(environment),
spawn,
createClient: (child) =>
this.deps._createCodexClient?.(child, this.logger, () => ({})) ??
new CodexAppServerClient(child, this.logger),
initializeParams: buildCodexAppServerInitializeParams(),
logger: this.logger,
});
try {
await client.request("initialize", buildCodexAppServerInitializeParams());
client.notify("initialized", {});
const limit = options?.limit ?? 20;
const scanLimit = Math.min(options?.scanLimit ?? limit, 500);
// thread/list returns the cheap `cwd` field. Fetch a wider window when
@@ -7749,15 +7819,10 @@ export class CodexAppServerAgentClient implements AgentClient {
try {
await runProviderRefreshActivity(context, "app-server.start", async () => {
const child = await this.spawnAppServer();
client = new CodexAppServerClient(child, this.logger);
client = await this.startAppServer({ signal: context?.signal });
if (context?.signal.aborted) await dispose();
});
if (!client) throw new Error("Codex app-server did not start");
await runProviderRefreshActivity(context, "initialize", () =>
client!.request("initialize", buildCodexAppServerInitializeParams()),
);
client.notify("initialized", {});
const rawResponse = await runProviderRefreshActivity(context, "model/list", () =>
client!.request("model/list", {}),
@@ -7801,12 +7866,9 @@ export class CodexAppServerAgentClient implements AgentClient {
const threadId = handle.nativeHandle ?? handle.sessionId;
if (!threadId) return;
const child = await this.spawnAppServer();
const client = new CodexAppServerClient(child, this.logger);
const client = await this.startAppServer();
try {
await client.request("initialize", buildCodexAppServerInitializeParams());
client.notify("initialized", {});
if (state === "archive") {
await client.request("thread/archive", { threadId });
return;
@@ -0,0 +1,347 @@
import { describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../../test-utils/test-logger.js";
import { createCodexAppServerChildProcess } from "./test-utils/fake-app-server.js";
import { CodexAppServerStartupCoordinator } from "./app-server-startup.js";
import { CodexAppServerExitError } from "./app-server-transport.js";
function sqliteStartupError(): CodexAppServerExitError {
return new CodexAppServerExitError({
exitCode: 1,
exitSignal: null,
stderr: "failed to initialize sqlite state runtime",
});
}
function deferred<T = void>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((resolvePromise) => {
resolve = resolvePromise;
});
return { promise, resolve };
}
describe("Codex app-server startup coordinator", () => {
test("returns initialize metadata so sessions retain server capabilities", async () => {
const coordinator = new CodexAppServerStartupCoordinator();
const client = {
request: async () => ({ userAgent: "codex/0.164.0" }),
notify: () => {},
dispose: async () => {},
};
const started = await coordinator.start({
stateDirKey: "/codex/metadata",
spawn: createCodexAppServerChildProcess,
createClient: () => client,
initializeParams: {},
logger: createTestLogger(),
});
expect(started).toEqual({ client, initialized: { userAgent: "codex/0.164.0" } });
});
test("serializes a burst through initialize but retains all initialized clients", async () => {
const coordinator = new CodexAppServerStartupCoordinator();
const firstEntered = deferred();
const releaseFirst = deferred();
let spawned = 0;
let activeInitializations = 0;
let maxActiveInitializations = 0;
let disposed = 0;
const starts = Array.from({ length: 24 }, () =>
coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
spawned++;
return createCodexAppServerChildProcess();
},
createClient: () => ({
request: async (method: string) => {
expect(method).toBe("initialize");
activeInitializations++;
maxActiveInitializations = Math.max(maxActiveInitializations, activeInitializations);
if (spawned === 1) {
firstEntered.resolve();
await releaseFirst.promise;
}
activeInitializations--;
return {};
},
notify: vi.fn(),
dispose: async () => {
disposed++;
},
}),
initializeParams: {},
logger: createTestLogger(),
}),
);
await firstEntered.promise;
expect(spawned).toBe(1);
releaseFirst.resolve();
const clients = await Promise.all(starts);
expect(clients).toHaveLength(24);
expect(spawned).toBe(24);
expect(maxActiveInitializations).toBe(1);
expect(disposed).toBe(0);
});
test("allows different Codex homes to initialize concurrently", async () => {
const coordinator = new CodexAppServerStartupCoordinator();
const bothEntered = deferred();
const release = deferred();
let entered = 0;
const start = (stateDirKey: string) =>
coordinator.start({
stateDirKey,
spawn: createCodexAppServerChildProcess,
createClient: () => ({
request: async () => {
entered++;
if (entered === 2) bothEntered.resolve();
await release.promise;
return {};
},
notify: vi.fn(),
dispose: vi.fn(async () => {}),
}),
initializeParams: {},
logger: createTestLogger(),
});
const first = start("/codex/one");
const second = start("/codex/two");
await bothEntered.promise;
expect(entered).toBe(2);
release.resolve();
await Promise.all([first, second]);
});
test("retries only SQLite startup failures after disposing the failed child", async () => {
const coordinator = new CodexAppServerStartupCoordinator();
const events: string[] = [];
let attempt = 0;
const client = await coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
attempt++;
events.push(`spawn ${attempt}`);
return createCodexAppServerChildProcess();
},
createClient: () => {
const number = attempt;
return {
request: async () => {
events.push(`initialize ${number}`);
if (number === 1) throw sqliteStartupError();
return {};
},
notify: () => events.push(`initialized ${number}`),
dispose: async () => {
events.push(`dispose ${number}`);
},
};
},
initializeParams: {},
logger: createTestLogger(),
});
expect(client.client).toEqual(expect.objectContaining({ request: expect.any(Function) }));
expect(events).toEqual([
"spawn 1",
"initialize 1",
"dispose 1",
"spawn 2",
"initialize 2",
"initialized 2",
]);
});
test("does not retry unrelated startup failures", async () => {
const coordinator = new CodexAppServerStartupCoordinator();
let spawned = 0;
let disposed = 0;
await expect(
coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
spawned++;
return createCodexAppServerChildProcess();
},
createClient: () => ({
request: async () => {
throw new Error("database is locked while reading unrelated configuration");
},
notify: vi.fn(),
dispose: async () => {
disposed++;
},
}),
initializeParams: {},
logger: createTestLogger(),
}),
).rejects.toThrow("database is locked while reading unrelated configuration");
expect(spawned).toBe(1);
expect(disposed).toBe(1);
});
test("bounds persistent SQLite failures and leaves the lane usable", async () => {
const coordinator = new CodexAppServerStartupCoordinator();
let spawned = 0;
let disposed = 0;
let locked = true;
const start = () =>
coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
spawned++;
return createCodexAppServerChildProcess();
},
createClient: () => ({
request: async () => {
if (locked) throw sqliteStartupError();
return {};
},
notify: vi.fn(),
dispose: async () => {
disposed++;
},
}),
initializeParams: {},
logger: createTestLogger(),
});
await expect(start()).rejects.toThrow("failed to initialize sqlite state runtime");
expect(spawned).toBe(3);
expect(disposed).toBe(3);
locked = false;
await expect(start()).resolves.toEqual(
expect.objectContaining({
client: expect.objectContaining({ request: expect.any(Function) }),
}),
);
expect(spawned).toBe(4);
expect(disposed).toBe(3);
});
test("a cancelled waiter never spawns or overtakes the active startup", async () => {
const coordinator = new CodexAppServerStartupCoordinator();
const firstEntered = deferred();
const releaseFirst = deferred();
const events: string[] = [];
const first = coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
events.push("spawn first");
return createCodexAppServerChildProcess();
},
createClient: () => ({
request: async () => {
firstEntered.resolve();
await releaseFirst.promise;
return {};
},
notify: vi.fn(),
dispose: vi.fn(async () => {}),
}),
initializeParams: {},
logger: createTestLogger(),
});
await firstEntered.promise;
const controller = new AbortController();
const queued = coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
events.push("spawn cancelled");
return createCodexAppServerChildProcess();
},
createClient: () => ({
request: async () => ({}),
notify: vi.fn(),
dispose: vi.fn(async () => {}),
}),
initializeParams: {},
logger: createTestLogger(),
signal: controller.signal,
});
controller.abort(new Error("cancelled"));
await expect(queued).rejects.toThrow("cancelled");
const next = coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
events.push("spawn next");
return createCodexAppServerChildProcess();
},
createClient: () => ({
request: async () => ({}),
notify: vi.fn(),
dispose: vi.fn(async () => {}),
}),
initializeParams: {},
logger: createTestLogger(),
});
expect(events).toEqual(["spawn first"]);
releaseFirst.resolve();
await Promise.all([first, next]);
expect(events).toEqual(["spawn first", "spawn next"]);
});
test("a timed-out initialize disposes its child and releases the next waiter", async () => {
vi.useFakeTimers();
try {
const coordinator = new CodexAppServerStartupCoordinator({
attemptTimeoutMs: 10,
queueTimeoutMs: 100,
});
const firstEntered = deferred();
const events: string[] = [];
const first = coordinator.start({
stateDirKey: "/codex/shared",
spawn: createCodexAppServerChildProcess,
createClient: () => ({
request: () => {
firstEntered.resolve();
return new Promise<never>(() => {});
},
notify: vi.fn(),
dispose: async () => {
events.push("dispose first");
},
}),
initializeParams: {},
logger: createTestLogger(),
});
await firstEntered.promise;
const next = coordinator.start({
stateDirKey: "/codex/shared",
spawn: () => {
events.push("spawn next");
return createCodexAppServerChildProcess();
},
createClient: () => ({
request: async () => ({}),
notify: vi.fn(),
dispose: vi.fn(async () => {}),
}),
initializeParams: {},
logger: createTestLogger(),
});
const firstResult = expect(first).rejects.toThrow("Timed out initializing Codex app-server");
await vi.advanceTimersByTimeAsync(10);
await firstResult;
await expect(next).resolves.toEqual(
expect.objectContaining({
client: expect.objectContaining({ request: expect.any(Function) }),
}),
);
expect(events).toEqual(["dispose first", "spawn next"]);
} finally {
vi.useRealTimers();
}
});
});
@@ -0,0 +1,189 @@
import type { ChildProcessWithoutNullStreams } from "node:child_process";
import type { Logger } from "pino";
import { terminateWithTreeKill } from "../../../../utils/tree-kill.js";
import { CodexAppServerExitError } from "./app-server-transport.js";
const DEFAULT_QUEUE_TIMEOUT_MS = 60_000;
const DEFAULT_ATTEMPT_TIMEOUT_MS = 30_000;
const DEFAULT_MAX_ATTEMPTS = 3;
interface InitializableAppServerClient {
request(method: string, params?: unknown, timeoutMs?: number): Promise<unknown>;
notify(method: string, params?: unknown): void;
dispose(): Promise<void>;
}
interface StartupOptions<TClient extends InitializableAppServerClient> {
stateDirKey: string;
spawn: () => ChildProcessWithoutNullStreams | Promise<ChildProcessWithoutNullStreams>;
createClient: (child: ChildProcessWithoutNullStreams) => TClient;
initializeParams: unknown;
logger: Pick<Logger, "warn" | "debug">;
signal?: AbortSignal;
}
interface StartedAppServer<TClient> {
client: TClient;
initialized: unknown;
}
interface CoordinatorOptions {
queueTimeoutMs?: number;
attemptTimeoutMs?: number;
maxAttempts?: number;
}
async function waitForSignal<T>(operation: Promise<T>, signal: AbortSignal): Promise<T> {
signal.throwIfAborted();
let onAbort: (() => void) | undefined;
const aborted = new Promise<never>((_resolve, reject) => {
onAbort = () => reject(signal.reason);
signal.addEventListener("abort", onAbort, { once: true });
});
try {
return await Promise.race([operation, aborted]);
} finally {
if (onAbort) signal.removeEventListener("abort", onAbort);
}
}
async function disposeUnownedChild(child: ChildProcessWithoutNullStreams): Promise<void> {
const result = await terminateWithTreeKill(child, {
gracefulTimeoutMs: 2_000,
forceTimeoutMs: 1_000,
});
if (result === "kill-timeout") {
throw new Error("Codex app-server did not report exit after SIGKILL");
}
}
export class CodexAppServerStartupCoordinator {
private readonly tails = new Map<string, Promise<void>>();
private readonly queueTimeoutMs: number;
private readonly attemptTimeoutMs: number;
private readonly maxAttempts: number;
constructor(options: CoordinatorOptions = {}) {
this.queueTimeoutMs = options.queueTimeoutMs ?? DEFAULT_QUEUE_TIMEOUT_MS;
this.attemptTimeoutMs = options.attemptTimeoutMs ?? DEFAULT_ATTEMPT_TIMEOUT_MS;
this.maxAttempts = options.maxAttempts ?? DEFAULT_MAX_ATTEMPTS;
}
async start<TClient extends InitializableAppServerClient>(
options: StartupOptions<TClient>,
): Promise<StartedAppServer<TClient>> {
options.signal?.throwIfAborted();
const previous = this.tails.get(options.stateDirKey) ?? Promise.resolve();
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
// An aborted waiter must not let later waiters pass the current owner.
const tail = previous.then(() => gate);
this.tails.set(options.stateDirKey, tail);
void tail.then(
() => this.tails.get(options.stateDirKey) === tail && this.tails.delete(options.stateDirKey),
);
const queueAbort = new AbortController();
const onAbort = () => queueAbort.abort(options.signal?.reason);
options.signal?.addEventListener("abort", onAbort, { once: true });
const queueTimer = setTimeout(
() => queueAbort.abort(new Error("Timed out waiting to start Codex app-server")),
this.queueTimeoutMs,
);
try {
await waitForSignal(previous, queueAbort.signal);
clearTimeout(queueTimer);
options.signal?.throwIfAborted();
for (let attempt = 1; attempt <= this.maxAttempts; attempt++) {
try {
return await this.startAttempt(options);
} catch (error) {
const isRetryableStartupFailure =
error instanceof CodexAppServerExitError &&
error.reason === "sqlite_state_initialization";
if (
options.signal?.aborted ||
attempt === this.maxAttempts ||
!isRetryableStartupFailure
) {
throw error;
}
options.logger.warn(
{ err: error, attempt, maxAttempts: this.maxAttempts },
"Codex app-server SQLite startup failed; retrying with a fresh process",
);
}
}
throw new Error("Codex app-server startup attempts exhausted");
} finally {
clearTimeout(queueTimer);
options.signal?.removeEventListener("abort", onAbort);
release();
}
}
private async startAttempt<TClient extends InitializableAppServerClient>(
options: StartupOptions<TClient>,
): Promise<StartedAppServer<TClient>> {
const attemptAbort = new AbortController();
const onAbort = () => attemptAbort.abort(options.signal?.reason);
options.signal?.addEventListener("abort", onAbort, { once: true });
const attemptTimer = setTimeout(
() => attemptAbort.abort(new Error("Timed out initializing Codex app-server")),
this.attemptTimeoutMs,
);
let child: ChildProcessWithoutNullStreams | undefined;
let client: TClient | undefined;
try {
attemptAbort.signal.throwIfAborted();
// Production hands in a synchronous spawn closure. Resolve the binary
// before entering the lane, then create the child while holding it.
const spawnPromise = Promise.resolve(options.spawn());
try {
child = await waitForSignal(spawnPromise, attemptAbort.signal);
} catch (error) {
if (attemptAbort.signal.aborted) {
// A future asynchronous spawner may finish after cancellation.
void spawnPromise
.then((lateChild) => disposeUnownedChild(lateChild))
.catch((cleanupError) =>
options.logger.warn({ err: cleanupError }, "Failed to clean up late Codex spawn"),
);
}
throw error;
}
attemptAbort.signal.throwIfAborted();
client = options.createClient(child);
const initialized = await waitForSignal(
client.request("initialize", options.initializeParams, this.attemptTimeoutMs),
attemptAbort.signal,
);
attemptAbort.signal.throwIfAborted();
client.notify("initialized", {});
return { client, initialized };
} catch (error) {
try {
if (client) {
await client.dispose();
} else if (child) {
await disposeUnownedChild(child);
}
} catch (cleanupError) {
options.logger.warn(
{ err: cleanupError, startupError: error },
"Failed to clean up Codex app-server after startup failure",
);
throw new Error("Codex app-server startup cleanup failed", { cause: cleanupError });
}
throw error;
} finally {
clearTimeout(attemptTimer);
options.signal?.removeEventListener("abort", onAbort);
}
}
}
export const codexAppServerStartup = new CodexAppServerStartupCoordinator();
@@ -5,9 +5,57 @@ import {
createCodexAppServerChildProcess,
createFakeCodexAppServer,
} from "./test-utils/fake-app-server.js";
import { CodexAppServerClient } from "./app-server-transport.js";
import { CodexAppServerClient, CodexAppServerExitError } from "./app-server-transport.js";
describe("Codex app-server transport", () => {
test("classifies startup diagnostics delivered after the child exits", async () => {
const child = createCodexAppServerChildProcess();
const client = new CodexAppServerClient(child, createTestLogger());
const request = client.request("initialize", {});
const result = expect(request).rejects.toMatchObject({
name: "CodexAppServerExitError",
reason: "sqlite_state_initialization",
stderr: "failed to initialize sqlite state runtime",
});
child.exitCode = 1;
child.emit("exit", 1, null);
child.stderr.write("failed to initialize sqlite state runtime");
child.emit("close", 1, null);
try {
await result;
} finally {
child.stdout.end();
child.stderr.end();
child.stdin.end();
}
});
test.each([
{
stderr: "Error: failed to initialize sqlite state runtime under /isolated/.codex",
reason: "sqlite_state_initialization",
},
{ stderr: "database is locked while loading configuration", reason: "other" },
] as const)("classifies a failed startup from child stderr as $reason", async (entry) => {
const child = createCodexAppServerChildProcess();
const client = new CodexAppServerClient(child, createTestLogger());
const request = client.request("initialize", {});
child.stderr.write(entry.stderr);
child.exitCode = 1;
child.emit("exit", 1, null);
await expect(request).rejects.toMatchObject({
name: "CodexAppServerExitError",
reason: entry.reason,
exitCode: 1,
exitSignal: null,
stderr: entry.stderr,
} satisfies Partial<CodexAppServerExitError>);
child.stdout.end();
child.stderr.end();
child.stdin.end();
});
test("ignores non-JSON stdout lines without dropping pending requests", async () => {
const child = createCodexAppServerChildProcess();
const client = new CodexAppServerClient(child, createTestLogger());
@@ -64,6 +64,39 @@ export class CodexAppServerRpcError extends Error {
}
}
interface CodexAppServerExitDetails {
exitCode: number | null;
exitSignal: NodeJS.Signals | null;
stderr: string;
}
export class CodexAppServerExitError extends Error {
readonly reason: "sqlite_state_initialization" | "other";
readonly exitCode: number | null;
readonly exitSignal: NodeJS.Signals | null;
readonly stderr: string;
constructor(details: CodexAppServerExitDetails) {
const message =
details.exitCode === 0 && !details.exitSignal
? "Codex app-server exited"
: `Codex app-server exited with code ${details.exitCode ?? "null"} and signal ${details.exitSignal ?? "null"}`;
super(`${message}\n${details.stderr}`.trim());
this.name = "CodexAppServerExitError";
this.exitCode = details.exitCode;
this.exitSignal = details.exitSignal;
this.stderr = details.stderr;
// Codex reports this pre-initialize failure only on stderr, not as a
// JSON-RPC error. Classify it here so callers branch on a typed reason.
this.reason =
details.exitCode === 1 &&
!details.exitSignal &&
details.stderr.includes("failed to initialize sqlite state runtime")
? "sqlite_state_initialization"
: "other";
}
}
type RequestHandler = (params: unknown, requestId: number) => unknown;
type NotificationHandler = (method: string, params: unknown) => void;
type UnexpectedTerminationHandler = (error: Error) => void;
@@ -234,14 +267,24 @@ export class CodexAppServerClient {
this.handleUnexpectedTermination(err);
});
const reportExit = (code: number | null, signal: NodeJS.Signals | null) => {
this.handleUnexpectedTermination(
new CodexAppServerExitError({
exitCode: code,
exitSignal: signal,
stderr: this.stderrBuffer,
}),
);
};
child.on("exit", (code, signal) => {
const message =
code === 0 && !signal
? "Codex app-server exited"
: `Codex app-server exited with code ${code ?? "null"} and signal ${signal ?? "null"}`;
const error = new Error(`${message}\n${this.stderrBuffer}`.trim());
this.handleUnexpectedTermination(error);
// Startup retries need complete diagnostics. Node drains the child's
// stdio before close, but exit can precede the final stderr data.
const initializing = Array.from(this.pending.values()).some(
(request) => request.method === "initialize",
);
if (!initializing) reportExit(code, signal);
});
child.on("close", reportExit);
}
setUnexpectedTerminationHandler(handler: UnexpectedTerminationHandler): void {
@@ -146,6 +146,10 @@ export function createCodexAppServerChildProcess(): CodexAppServerChildProcess {
exitCode: null,
signalCode: null,
}) as CodexAppServerChildProcess;
// A real ChildProcess closes its stdio after exit, then emits close.
child.once("exit", (code, signal) => {
queueMicrotask(() => child.emit("close", code, signal));
});
child.kill = ((signal?: NodeJS.Signals | number) => {
queueMicrotask(() => child.emit("exit", null, signal ?? null));
return true;
@@ -1,4 +1,3 @@
import type { ChildProcessWithoutNullStreams } from "node:child_process";
import { describe, expect, test } from "vitest";
import { createTestLogger } from "../../../../test-utils/test-logger.js";
@@ -27,15 +26,24 @@ async function openCodexThread(
const appServer = createFakeCodexAppServer(
fakeCodexThreadHistory({ "saved-thread": thread }, options),
);
const provider = new CodexAppServerAgentClient(createTestLogger());
const provider = new CodexAppServerAgentClient(
createTestLogger(),
{
command: { mode: "replace", argv: ["codex-must-not-run-in-this-test"] },
},
{
appServerProcess: {
resolvePrefix: async () => ({ command: "codex", args: [] }),
spawn: () => appServer.child,
},
},
);
const internals = provider as unknown as {
goalsEnabledPromise: Promise<boolean> | null;
autoReviewEnabledPromise: Promise<boolean> | null;
spawnAppServer: () => Promise<ChildProcessWithoutNullStreams>;
};
internals.goalsEnabledPromise = Promise.resolve(false);
internals.autoReviewEnabledPromise = Promise.resolve(false);
internals.spawnAppServer = async () => appServer.child;
const session = await provider.resumeSession({ sessionId: "saved-thread" });
return { appServer, session };
}