fix(codex): bound shared app-server startup waits (#89442)

* 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 <zhang.guiping@xydigit.com>

* 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 <steipete@gmail.com>

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<typeof startCodexAttemptThread>[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<void> {
@@ -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<void>((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<void>((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 () => {

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;

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<void> => 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<void>((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();

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<CodexAppServerClient>;
+  startup?: SharedCodexAppServerClientStartup;
   activeLeases: number;
   pendingAcquires: number;
   closeWhenIdle: boolean;
   closeError?: Error;
 };
 
+type SharedCodexAppServerClientStartup = {
+  initialized: Promise<void>;
+  ready: Promise<CodexAppServerClient>;
+};
+
 type SharedCodexAppServerClientState = {
   clients: Map<string, SharedCodexAppServerClientEntry>;
   leasedReleases: WeakMap<CodexAppServerClient, Array<() => 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<void>();
+  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<CodexAppServerClient> {
   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);