From 745cfc028edb2d659c382d88629a70b1ee2a45fc Mon Sep 17 00:00:00 2001 From: xingzhou Date: Sat, 11 Jul 2026 19:58:55 +0800 Subject: [PATCH] fix(codex): bound shared app-server startup waits (#89442) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(codex): isolated cron reports Codex startup stalls * fix(codex): isolated cron reports Codex startup stalls * fix(codex): bound shared app-server startup waits Co-authored-by: 张贵萍0668001030 * test(codex): satisfy startup lifecycle checks * test(codex): align deferred auth test type * test(codex): type auth mock as async void --------- Co-authored-by: Peter Steinberger --- .../src/app-server/attempt-startup.test.ts | 160 ++++++++++++++++-- .../codex/src/app-server/attempt-startup.ts | 18 +- .../src/app-server/shared-client.test.ts | 63 ++++++- .../codex/src/app-server/shared-client.ts | 102 +++++++---- 4 files changed, 298 insertions(+), 45 deletions(-) diff --git a/extensions/codex/src/app-server/attempt-startup.test.ts b/extensions/codex/src/app-server/attempt-startup.test.ts index 51a0dcdff925..0e1df549952c 100644 --- a/extensions/codex/src/app-server/attempt-startup.test.ts +++ b/extensions/codex/src/app-server/attempt-startup.test.ts @@ -14,8 +14,10 @@ import { type CodexPluginConfig, resolveCodexAppServerRuntimeOptions } from "./c import { testCodexAppServerBindingStore } from "./session-binding.test-helpers.js"; import { clearSharedCodexAppServerClient, + clearSharedCodexAppServerClientAndWait, getLeasedSharedCodexAppServerClient, releaseLeasedSharedCodexAppServerClient, + type CodexAppServerClientFactory, } from "./shared-client.js"; import { createClientHarness, createCodexTestModel } from "./test-support.js"; @@ -85,9 +87,8 @@ function startThreadWithHarness( signal = new AbortController().signal, overrides?: { pluginConfig?: CodexPluginConfig; - attemptClientFactory?: ( - harness: ClientHarness, - ) => Parameters[0]["attemptClientFactory"]; + attemptClientFactory?: (harness: ClientHarness) => CodexAppServerClientFactory; + buildAttemptParams?: () => EmbeddedRunAttemptParams; harness?: ClientHarness; paths?: AttemptPaths; skipStartSpy?: boolean; @@ -95,9 +96,9 @@ function startThreadWithHarness( ) { const harness = overrides?.harness ?? createClientHarness(); const paths = overrides?.paths ?? createAttemptPaths(); - if (!overrides?.skipStartSpy) { - vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client); - } + const startSpy = overrides?.skipStartSpy + ? undefined + : vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client); const effectivePluginConfig = overrides?.pluginConfig ?? pluginConfig; const run = startCodexAttemptThread({ @@ -112,7 +113,7 @@ function startThreadWithHarness( startupEnvApiKeyCacheKey: undefined, agentDir: paths.agentDir, config: undefined, - buildAttemptParams: () => createAttemptParams(paths), + buildAttemptParams: overrides?.buildAttemptParams ?? (() => createAttemptParams(paths)), sessionAgentId: "agent-1", effectiveWorkspace: paths.workspaceDir, effectiveCwd: paths.cwd, @@ -132,7 +133,7 @@ function startThreadWithHarness( spawnedBy: undefined, }); - return { harness, run }; + return { harness, run, startSpy }; } async function answerInitialize(harness: ClientHarness): Promise { @@ -166,6 +167,15 @@ async function waitForThreadStart(harness: ClientHarness): Promise<{ id?: number return waitForRequest(harness, "thread/start"); } +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + describe("startCodexAttemptThread", () => { beforeEach(() => { vi.useRealTimers(); @@ -291,7 +301,13 @@ describe("startCodexAttemptThread", () => { }); it("closes the shared app-server when startup times out during initialize", async () => { - const { harness, run } = startThreadWithHarness(500); + const initializeTimeoutPluginConfig = { + ...pluginConfig, + appServer: { command: "codex", requestTimeoutMs: 100 }, + } satisfies CodexPluginConfig; + const { harness, run } = startThreadWithHarness(500, new AbortController().signal, { + pluginConfig: initializeTimeoutPluginConfig, + }); const runError = run.then( () => undefined, (error: unknown) => error, @@ -302,7 +318,7 @@ describe("startCodexAttemptThread", () => { const error = await runError; expect(error).toBeInstanceOf(Error); - expect((error as Error).message).toBe("codex app-server startup timed out"); + expect((error as Error).message).toBe("codex app-server initialize timed out"); await vi.waitFor(() => expect(harness.stdinDestroyed).toBe(true), { interval: 1, timeout: 1_000, @@ -312,20 +328,142 @@ describe("startCodexAttemptThread", () => { ).toBe(false); }); + it("does not retire shared startup when this attempt's initialize wait expires", async () => { + const sharedInitializePluginConfig = { + ...pluginConfig, + appServer: { command: "codex", requestTimeoutMs: 100 }, + } satisfies CodexPluginConfig; + const appServer = resolveCodexAppServerRuntimeOptions({ + pluginConfig: sharedInitializePluginConfig, + }); + const paths = createAttemptPaths(); + const { harness, run, startSpy } = startThreadWithHarness(1_000, new AbortController().signal, { + pluginConfig: sharedInitializePluginConfig, + paths, + }); + await waitForRequest(harness, "initialize"); + const peerAcquire = getLeasedSharedCodexAppServerClient({ + startOptions: appServer.start, + agentDir: paths.agentDir, + timeoutMs: 1_000, + }); + + await expect(run).rejects.toThrow("codex app-server initialize timed out"); + expect(harness.stdinDestroyed).toBe(false); + await answerInitialize(harness); + await expect(peerAcquire).resolves.toBe(harness.client); + await expect( + getLeasedSharedCodexAppServerClient({ + startOptions: appServer.start, + agentDir: paths.agentDir, + timeoutMs: 1_000, + }), + ).resolves.toBe(harness.client); + expect(startSpy).toHaveBeenCalledTimes(1); + expect(releaseLeasedSharedCodexAppServerClient(harness.client)).toBe(true); + expect(releaseLeasedSharedCodexAppServerClient(harness.client)).toBe(true); + }); + + it("bounds a real stdio initialize request and cleans up the child", async () => { + const paths = createAttemptPaths(); + const root = path.dirname(paths.agentDir); + const fixturePath = path.join(root, "stall-initialize.mjs"); + const requestLogPath = path.join(root, "requests.log"); + const pidPath = path.join(root, "child.pid"); + await fs.mkdir(root, { recursive: true }); + await fs.writeFile( + fixturePath, + [ + 'import fs from "node:fs";', + 'import readline from "node:readline";', + "const [requestLogPath, pidPath] = process.argv.slice(2);", + 'fs.writeFileSync(pidPath, String(process.pid), "utf8");', + "const lines = readline.createInterface({ input: process.stdin });", + 'lines.on("line", (line) => {', + " const message = JSON.parse(line);", + ' fs.appendFileSync(requestLogPath, `${String(message.method)}\\n`, "utf8");', + "});", + "setInterval(() => undefined, 1000);", + ].join("\n"), + "utf8", + ); + const stdioPluginConfig = { + appServer: { + transport: "stdio", + command: process.execPath, + args: [fixturePath, requestLogPath, pidPath], + requestTimeoutMs: 2_000, + }, + } satisfies CodexPluginConfig; + let childPid: number | undefined; + + try { + const { run } = startThreadWithHarness(5_000, new AbortController().signal, { + pluginConfig: stdioPluginConfig, + paths, + skipStartSpy: true, + }); + + await expect(run).rejects.toThrow("codex app-server initialize timed out"); + + const requestMethods = (await fs.readFile(requestLogPath, "utf8")).trim().split(/\r?\n/u); + expect(requestMethods).toEqual(["initialize"]); + childPid = Number.parseInt(await fs.readFile(pidPath, "utf8"), 10); + expect(childPid).toBeGreaterThan(0); + const observedPid = childPid; + await vi.waitFor(() => expect(isProcessAlive(observedPid)).toBe(false), { + interval: 25, + timeout: 3_000, + }); + } finally { + await clearSharedCodexAppServerClientAndWait({ + exitTimeoutMs: 3_000, + forceKillDelayMs: 100, + }); + if (childPid && isProcessAlive(childPid)) { + try { + process.kill(childPid, "SIGKILL"); + } catch { + // The child can exit between the liveness probe and fallback kill. + } + } + } + }); + + it("cleans up a client surfaced by a factory that later rejects", async () => { + const { harness, run } = startThreadWithHarness(5_000, new AbortController().signal, { + attemptClientFactory: (factoryHarness) => async (options) => { + options?.onStartedClient?.(factoryHarness.client); + throw new Error("custom initialize failed"); + }, + }); + + await expect(run).rejects.toThrow("custom initialize failed"); + expect(harness.stdinDestroyed).toBe(true); + }); + it("closes a startup client that arrives after startup timeout", async () => { let observedFactoryOptions: | { onStartedClient?: (client: CodexAppServerClient) => void; abandonSignal?: AbortSignal; + timeoutMs?: number; } | undefined; + let factoryCalls = 0; let resolveFactoryDone: () => void = () => undefined; const factoryDone = new Promise((resolve) => { resolveFactoryDone = resolve; }); + const delayedFactoryPluginConfig = { + ...pluginConfig, + appServer: { command: "codex", requestTimeoutMs: 2_500 }, + } satisfies CodexPluginConfig; const { harness, run } = startThreadWithHarness(100, new AbortController().signal, { + pluginConfig: delayedFactoryPluginConfig, attemptClientFactory: (factoryHarness) => async (options) => { try { + factoryCalls += 1; observedFactoryOptions = options; await new Promise((resolve) => { setTimeout(resolve, 250); @@ -350,6 +488,8 @@ describe("startCodexAttemptThread", () => { ).toBe(false); expect(observedFactoryOptions?.onStartedClient).toBeTypeOf("function"); expect(observedFactoryOptions?.abandonSignal?.aborted).toBe(true); + expect(observedFactoryOptions?.timeoutMs).toBe(2_500); + expect(factoryCalls).toBe(1); }); it("clears the shared app-server when cancellation abandons an in-flight thread request", async () => { diff --git a/extensions/codex/src/app-server/attempt-startup.ts b/extensions/codex/src/app-server/attempt-startup.ts index 6451aaffb52b..a68e35b67469 100644 --- a/extensions/codex/src/app-server/attempt-startup.ts +++ b/extensions/codex/src/app-server/attempt-startup.ts @@ -60,6 +60,7 @@ import { import type { CodexAppServerBindingStore } from "./session-binding.js"; import { clearSharedCodexAppServerClientIfCurrent, + clearSharedCodexAppServerClientIfCurrentAndUnclaimed, releaseLeasedSharedCodexAppServerClient, type CodexAppServerClientFactory, } from "./shared-client.js"; @@ -206,6 +207,7 @@ export async function startCodexAttemptThread(params: { } }, abandonSignal: startupAbandonController.signal, + timeoutMs: params.appServer.requestTimeoutMs, }); const activeStartupClient = startupClient; let startupClientLeaseReleased = false; @@ -455,6 +457,16 @@ export async function startCodexAttemptThread(params: { } } catch (error) { startupAttemptError = error; + if (!startupAbandoned && !params.signal.aborted && !startupClient) { + const sharedClient = clearSharedCodexAppServerClientIfCurrentAndUnclaimed( + startupClientForAbandonedRequestCleanup, + ); + if (sharedClient.found && !sharedClient.closed) { + // Shared acquisition already released this caller. A peer still + // owns the client, so outer cleanup must not retire it. + startupClientForAbandonedRequestCleanup = undefined; + } + } throw error; } finally { if (!startupAttemptSucceeded) { @@ -491,7 +503,11 @@ export async function startCodexAttemptThread(params: { try { return await startupAttempt(); } catch (error) { - if (params.signal.aborted || !isCodexAppServerConnectionClosedError(error)) { + if ( + startupAbandoned || + params.signal.aborted || + !isCodexAppServerConnectionClosedError(error) + ) { throw error; } const failedClient = attemptedClient; diff --git a/extensions/codex/src/app-server/shared-client.test.ts b/extensions/codex/src/app-server/shared-client.test.ts index 7d2d87d25d91..c463cedd0d84 100644 --- a/extensions/codex/src/app-server/shared-client.test.ts +++ b/extensions/codex/src/app-server/shared-client.test.ts @@ -7,7 +7,11 @@ import { createClientHarness } from "./test-support.js"; const mocks = vi.hoisted(() => ({ bridgeCodexAppServerStartOptions: vi.fn(async ({ startOptions }) => startOptions), applyCodexAppServerAuthProfile: vi.fn( - async (_params?: { agentDir?: string; authProfileId?: string; config?: unknown }) => undefined, + async (_params?: { + agentDir?: string; + authProfileId?: string; + config?: unknown; + }): Promise => undefined, ), resolveCodexAppServerAuthProfileIdForAgent: vi.fn( (params?: { authProfileId?: string }) => params?.authProfileId, @@ -126,6 +130,15 @@ function clientStartCall(startSpy: unknown) { }; } +function deferNextAuthProfileApplication(): () => void { + let release: () => void = () => {}; + const gate = new Promise((resolve) => { + release = () => resolve(); + }); + mocks.applyCodexAppServerAuthProfile.mockReturnValueOnce(gate); + return release; +} + describe("shared Codex app-server client", () => { beforeAll(async () => { ({ listCodexAppServerModels } = await import("./models.js")); @@ -242,6 +255,54 @@ describe("shared Codex app-server client", () => { expect(startSpy).toHaveBeenCalledTimes(2); }); + it("keeps shared startup alive for a caller with a longer initialize timeout", async () => { + const harness = createClientHarness(); + const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client); + + const shortAcquire = getSharedCodexAppServerClient({ timeoutMs: 5 }); + const longAcquire = getSharedCodexAppServerClient({ timeoutMs: 1000 }); + + await expect(shortAcquire).rejects.toThrow("codex app-server initialize timed out"); + expect(harness.process.stdin.destroyed).toBe(false); + + await sendInitializeResult(harness, "openclaw/0.143.0 (macOS; test)"); + + await expect(longAcquire).resolves.toBe(harness.client); + expect(startSpy).toHaveBeenCalledTimes(1); + expect(harness.process.stdin.destroyed).toBe(false); + }); + + it("reports a stalled shared auth phase separately from initialize", async () => { + const harness = createClientHarness(); + vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client); + const releaseAuth = deferNextAuthProfileApplication(); + + const acquire = getSharedCodexAppServerClient({ timeoutMs: 100 }); + await sendInitializeResult(harness, "openclaw/0.143.0 (macOS; test)"); + + await expect(acquire).rejects.toThrow("codex app-server authentication timed out"); + expect(harness.process.stdin.destroyed).toBe(true); + releaseAuth(); + }); + + it("keeps shared auth alive for a caller with a longer timeout", async () => { + const harness = createClientHarness(); + const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client); + const releaseAuth = deferNextAuthProfileApplication(); + + const shortAcquire = getSharedCodexAppServerClient({ timeoutMs: 100 }); + const longAcquire = getSharedCodexAppServerClient({ timeoutMs: 1000 }); + await sendInitializeResult(harness, "openclaw/0.143.0 (macOS; test)"); + + await expect(shortAcquire).rejects.toThrow("codex app-server authentication timed out"); + expect(harness.process.stdin.destroyed).toBe(false); + + releaseAuth(); + await expect(longAcquire).resolves.toBe(harness.client); + expect(startSpy).toHaveBeenCalledTimes(1); + expect(harness.process.stdin.destroyed).toBe(false); + }); + it("keeps a pending shared app-server alive when another acquire still owns startup", async () => { const harness = createClientHarness(); const abandonController = new AbortController(); diff --git a/extensions/codex/src/app-server/shared-client.ts b/extensions/codex/src/app-server/shared-client.ts index 9231292f976c..8d9bb588b911 100644 --- a/extensions/codex/src/app-server/shared-client.ts +++ b/extensions/codex/src/app-server/shared-client.ts @@ -3,6 +3,7 @@ * lease tracking, and teardown. */ import { resolveDefaultAgentDir, type AuthProfileStore } from "openclaw/plugin-sdk/agent-runtime"; +import { createDeferred } from "openclaw/plugin-sdk/extension-shared"; import { applyCodexAppServerAuthProfile, bridgeCodexAppServerStartOptions, @@ -22,13 +23,18 @@ import { withTimeout } from "./timeout.js"; type SharedCodexAppServerClientEntry = { client?: CodexAppServerClient; - promise?: Promise; + startup?: SharedCodexAppServerClientStartup; activeLeases: number; pendingAcquires: number; closeWhenIdle: boolean; closeError?: Error; }; +type SharedCodexAppServerClientStartup = { + initialized: Promise; + ready: Promise; +}; + type SharedCodexAppServerClientState = { clients: Map; leasedReleases: WeakMap void>>; @@ -194,29 +200,28 @@ async function acquireSharedCodexAppServerClient( abandon(); } } - const sharedPromise = - entry.promise ?? - (entry.promise = (async () => { - const client = await startInitializedCodexAppServerClient({ - startOptions, - agentDir, - authProfileId: usesNativeAuth ? null : authProfileId, - config: options?.config, - onStartedClient: (startedClient) => { - entry.client = startedClient; - options?.onStartedClient?.(startedClient); - }, - }); - entry.client = client; - client.addCloseHandler((closedClient) => clearSharedClientEntryIfCurrent(key, closedClient)); - return client; - })()); + const startup = + entry.startup ?? + (entry.startup = createSharedCodexAppServerClientStartup({ + entry, + key, + startOptions, + agentDir, + authProfileId: usesNativeAuth ? null : authProfileId, + config: options?.config, + onStartedClient: options?.onStartedClient, + })); try { - const client = await withTimeout( - sharedPromise, + await withTimeout( + startup.initialized, options?.timeoutMs ?? 0, "codex app-server initialize timed out", ); + const client = await withTimeout( + startup.ready, + options?.timeoutMs ?? 0, + "codex app-server authentication timed out", + ); if (entry.closeError) { throw entry.closeError; } @@ -230,10 +235,10 @@ async function acquireSharedCodexAppServerClient( const release = leaseOptions?.leased ? retainSharedClientEntry(entry) : undefined; return release ? { client, release } : { client }; } catch (error) { - const currentEntry = state.clients.get(key); - if (currentEntry?.promise === sharedPromise) { - clearSharedClientEntry(key, currentEntry); - } + // This deadline belongs to one waiter, not the shared physical client. + // Release first so only the final claimant can tear down stalled startup. + releasePendingAcquire(); + closeSharedClientEntryIfUnclaimed(key, entry); throw error; } finally { cleanupAbandonSignal?.(); @@ -241,6 +246,44 @@ async function acquireSharedCodexAppServerClient( } } +function createSharedCodexAppServerClientStartup(params: { + entry: SharedCodexAppServerClientEntry; + key: string; + startOptions: CodexAppServerStartOptions; + agentDir: string; + authProfileId: string | null | undefined; + config?: CodexAppServerClientOptions["config"]; + onStartedClient?: (client: CodexAppServerClient) => void; +}): SharedCodexAppServerClientStartup { + const initialized = createDeferred(); + const ready = startInitializedCodexAppServerClient({ + startOptions: params.startOptions, + agentDir: params.agentDir, + authProfileId: params.authProfileId, + config: params.config, + onStartedClient: (startedClient) => { + params.entry.client = startedClient; + params.onStartedClient?.(startedClient); + }, + onInitializedClient: () => initialized.resolve(), + }).then( + (client) => { + params.entry.client = client; + client.addCloseHandler((closedClient) => + clearSharedClientEntryIfCurrent(params.key, closedClient), + ); + return client; + }, + (error: unknown) => { + initialized.reject(error); + throw error; + }, + ); + // Callers observe pre-initialize failures through the phase promise first. + void ready.catch(() => undefined); + return { initialized: initialized.promise, ready }; +} + /** Starts a non-shared Codex app-server client owned entirely by the caller. */ export async function createIsolatedCodexAppServerClient( options?: IsolatedCodexAppServerClientOptions, @@ -266,6 +309,7 @@ async function startInitializedCodexAppServerClient(params: { config?: CodexAppServerClientOptions["config"]; timeoutMs?: number; onStartedClient?: (client: CodexAppServerClient) => void; + onInitializedClient?: () => void; }): Promise { const startOptionsCandidates = resolveManagedFallbackStartOptions(params.startOptions); for (let index = 0; index < startOptionsCandidates.length; index += 1) { @@ -284,6 +328,7 @@ async function startInitializedCodexAppServerClient(params: { throw error; } + params.onInitializedClient?.(); ensureCodexAppServerClientRuntime(client, { agentDir: params.agentDir, authProfileId: params.authProfileId ?? undefined, @@ -509,15 +554,6 @@ function getOrCreateSharedClientEntry( return entry; } -function clearSharedClientEntry(key: string, entry: SharedCodexAppServerClientEntry): void { - const state = getSharedCodexAppServerClientState(); - if (state.clients.get(key) !== entry) { - return; - } - state.clients.delete(key); - entry.client?.close(); -} - function clearSharedClientEntryIfCurrent(key: string, client: CodexAppServerClient): void { const state = getSharedCodexAppServerClientState(); const entry = state.clients.get(key);