fix(codex): prevent aborted app-server turn handles

docs/plugins/codex-harness-reference.md

@@ -368,7 +368,7 @@ If discovery fails or times out, OpenClaw uses a bundled fallback catalog for:
 - GPT-5.4 mini
 - GPT-5.2
 
-The current bundled harness is `@openai/codex` `0.134.0`. A `model/list` probe
+The current bundled harness is `@openai/codex` `0.135.0`. A `model/list` probe
 against that bundled app-server returned:
 
 | Model id              | Default | Hidden | Input modalities | Reasoning efforts        |

extensions/codex/src/app-server/attempt-startup.test.ts

@@ -121,12 +121,14 @@ async function waitForThreadStart(harness: ClientHarness): Promise<{ id?: number
 
 describe("startCodexAttemptThread", () => {
   beforeEach(() => {
+    vi.useRealTimers();
     vi.stubEnv("CODEX_API_KEY", "");
     vi.stubEnv("OPENAI_API_KEY", "");
     clearSharedCodexAppServerClient();
   });
 
   afterEach(() => {
+    vi.useRealTimers();
     clearSharedCodexAppServerClient();
     vi.restoreAllMocks();
     vi.unstubAllEnvs();

extensions/codex/src/app-server/attempt-steering.test.ts

@@ -1,7 +1,11 @@
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { createCodexSteeringQueue } from "./attempt-steering.js";
 
 describe("Codex app-server steering queue", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
   afterEach(() => {
     vi.useRealTimers();
   });
@@ -16,7 +20,9 @@ describe("Codex app-server steering queue", () => {
       signal: new AbortController().signal,
     });
 
-    await queue.queue("accepted", { debounceMs: 0 });
+    const queued = queue.queue("accepted", { debounceMs: 0 });
+    await vi.advanceTimersByTimeAsync(0);
+    await queued;
 
     expect(request).toHaveBeenCalledWith("turn/steer", {
       threadId: "thread-1",
@@ -37,9 +43,10 @@ describe("Codex app-server steering queue", () => {
       signal: new AbortController().signal,
     });
 
-    await expect(queue.queue("rejected", { debounceMs: 0 })).rejects.toThrow(
-      "cannot steer a compact turn",
-    );
+    const queued = queue.queue("rejected", { debounceMs: 0 });
+    const rejected = expect(queued).rejects.toThrow("cannot steer a compact turn");
+    await vi.advanceTimersByTimeAsync(0);
+    await rejected;
     expect(request).toHaveBeenCalledWith("turn/steer", {
       threadId: "thread-1",
       expectedTurnId: "turn-1",
@@ -48,7 +55,6 @@ describe("Codex app-server steering queue", () => {
   });
 
   it("rejects queued steering when the run aborts before debounce flush", async () => {
-    vi.useFakeTimers();
     const controller = new AbortController();
     const request = vi.fn(async () => ({ turnId: "turn-1" }));
     const queue = createCodexSteeringQueue({

extensions/codex/src/app-server/auth-profile-runtime-contract.test.ts

@@ -157,10 +157,12 @@ describe("Auth profile runtime contract - Codex app-server adapter", () => {
   let tmpDir: string;
 
   beforeEach(async () => {
+    vi.useRealTimers();
     tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-auth-contract-"));
   });
 
   afterEach(async () => {
+    vi.useRealTimers();
     abortAgentHarnessRun(AUTH_PROFILE_RUNTIME_CONTRACT.sessionId);
     resetCodexAppServerClientFactoryForTest();
     await fs.rm(tmpDir, { recursive: true, force: true });

extensions/codex/src/app-server/run-attempt-test-harness.ts

@@ -465,6 +465,7 @@ export function createRuntimeDynamicTool(name: string): RuntimeDynamicToolForTes
 
 export function setupRunAttemptTestHooks(): void {
   beforeEach(async () => {
+    vi.useRealTimers();
     clearInternalHooks();
     resetAgentEventsForTest();
     resetDiagnosticEventsForTest();

extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts

@@ -84,6 +84,7 @@ function turnStartResult(turnId = "turn-1") {
 
 describe("Codex app-server main thread cleanup", () => {
   beforeEach(async () => {
+    vi.useRealTimers();
     resetAgentEventsForTest();
     vi.stubEnv("OPENCLAW_TRAJECTORY", "0");
     vi.stubEnv("CODEX_API_KEY", "");
@@ -92,6 +93,7 @@ describe("Codex app-server main thread cleanup", () => {
   });
 
   afterEach(async () => {
+    vi.useRealTimers();
     resetAgentEventsForTest();
     vi.restoreAllMocks();
     vi.unstubAllEnvs();

extensions/codex/src/app-server/run-attempt.steering.test.ts

@@ -18,24 +18,44 @@ import {
 
 setupRunAttemptTestHooks();
 
+function createSteeringParams(name: string) {
+  const params = createParams(
+    path.join(tempDir, `${name}.jsonl`),
+    path.join(tempDir, `${name}-workspace`),
+  );
+  params.sessionId = `session-${name}`;
+  params.sessionKey = `agent:main:session-${name}`;
+  return params;
+}
+
+async function queueActiveRunMessageEventually(
+  sessionId: string,
+  text: string,
+  options?: Parameters<typeof queueActiveRunMessageForTest>[2],
+) {
+  await vi.waitFor(
+    () => expect(queueActiveRunMessageForTest(sessionId, text, options)).toBe(true),
+    fastWait,
+  );
+}
+
 describe("runCodexAppServerAttempt steering", () => {
   it("forwards queued user input and aborts the active app-server turn", async () => {
     const { requests, waitForMethod } = createStartedThreadHarness();
+    const params = createSteeringParams("steering-forward");
 
-    const run = runCodexAppServerAttempt(
-      createParams(path.join(tempDir, "session.jsonl"), path.join(tempDir, "workspace")),
-      { pluginConfig: { appServer: { mode: "yolo" } } },
-    );
+    const run = runCodexAppServerAttempt(params, { pluginConfig: { appServer: { mode: "yolo" } } });
     await waitForMethod("turn/start");
 
-    expect(queueActiveRunMessageForTest("session-1", "more context", { debounceMs: 1 })).toBe(true);
-    await vi.waitFor(() => expect(requests.map((entry) => entry.method)).toContain("turn/steer"), {
-      interval: 1,
-    });
-    expect(abortAgentHarnessRun("session-1")).toBe(true);
+    await queueActiveRunMessageEventually(params.sessionId, "more context", { debounceMs: 1 });
+    await vi.waitFor(
+      () => expect(requests.map((entry) => entry.method)).toContain("turn/steer"),
+      fastWait,
+    );
+    expect(abortAgentHarnessRun(params.sessionId)).toBe(true);
     await vi.waitFor(
       () => expect(requests.map((entry) => entry.method)).toContain("turn/interrupt"),
-      { interval: 1 },
+      fastWait,
     );
 
     const result = await run;
@@ -67,22 +87,21 @@ describe("runCodexAppServerAttempt steering", () => {
 
   it("accepts message-tool-only steering for active Codex app-server source replies", async () => {
     const { requests, waitForMethod, completeTurn } = createStartedThreadHarness();
-    const params = createParams(
-      path.join(tempDir, "session.jsonl"),
-      path.join(tempDir, "workspace"),
-    );
+    const params = createSteeringParams("steering-message-tool");
     params.sourceReplyDeliveryMode = "message_tool_only";
 
     const run = runCodexAppServerAttempt(params);
     await waitForMethod("turn/start");
 
-    expect(
-      queueActiveRunMessageForTest("session-1", "subagent complete", {
+    await queueActiveRunMessageEventually(
+      params.sessionId,
+      "subagent complete",
+      {
         debounceMs: 1,
         steeringMode: "all",
         sourceReplyDeliveryMode: "message_tool_only",
-      }),
-    ).toBe(true);
+      },
+    );
 
     await vi.waitFor(
       () =>
@@ -96,7 +115,7 @@ describe("runCodexAppServerAttempt steering", () => {
             },
           },
         ]),
-      { interval: 1 },
+      fastWait,
     );
 
     await completeTurn({ threadId: "thread-1", turnId: "turn-1" });
@@ -105,14 +124,13 @@ describe("runCodexAppServerAttempt steering", () => {
 
   it("batches default queued steering before sending turn/steer", async () => {
     const { requests, waitForMethod, completeTurn } = createStartedThreadHarness();
+    const params = createSteeringParams("steering-batch-default");
 
-    const run = runCodexAppServerAttempt(
-      createParams(path.join(tempDir, "session.jsonl"), path.join(tempDir, "workspace")),
-    );
+    const run = runCodexAppServerAttempt(params);
     await waitForMethod("turn/start");
 
-    expect(queueActiveRunMessageForTest("session-1", "first", { debounceMs: 5 })).toBe(true);
-    expect(queueActiveRunMessageForTest("session-1", "second", { debounceMs: 5 })).toBe(true);
+    await queueActiveRunMessageEventually(params.sessionId, "first", { debounceMs: 5 });
+    expect(queueActiveRunMessageForTest(params.sessionId, "second", { debounceMs: 5 })).toBe(true);
 
     await vi.waitFor(
       () =>
@@ -129,7 +147,7 @@ describe("runCodexAppServerAttempt steering", () => {
             },
           },
         ]),
-      { interval: 1 },
+      fastWait,
     );
 
     await completeTurn({ threadId: "thread-1", turnId: "turn-1" });
@@ -138,15 +156,12 @@ describe("runCodexAppServerAttempt steering", () => {
 
   it("flushes pending default queued steering during normal turn cleanup", async () => {
     const { requests, waitForMethod, completeTurn } = createStartedThreadHarness();
+    const params = createSteeringParams("steering-flush");
 
-    const run = runCodexAppServerAttempt(
-      createParams(path.join(tempDir, "session.jsonl"), path.join(tempDir, "workspace")),
-    );
+    const run = runCodexAppServerAttempt(params);
     await waitForMethod("turn/start");
 
-    expect(queueActiveRunMessageForTest("session-1", "late steer", { debounceMs: 30_000 })).toBe(
-      true,
-    );
+    await queueActiveRunMessageEventually(params.sessionId, "late steer", { debounceMs: 30_000 });
 
     await completeTurn({ threadId: "thread-1", turnId: "turn-1" });
     await run;
@@ -165,17 +180,20 @@ describe("runCodexAppServerAttempt steering", () => {
 
   it("batches explicit all-mode steering before sending turn/steer", async () => {
     const { requests, waitForMethod, completeTurn } = createStartedThreadHarness();
+    const params = createSteeringParams("steering-batch-all");
 
-    const run = runCodexAppServerAttempt(
-      createParams(path.join(tempDir, "session.jsonl"), path.join(tempDir, "workspace")),
-    );
+    const run = runCodexAppServerAttempt(params);
     await waitForMethod("turn/start");
 
+    await queueActiveRunMessageEventually(params.sessionId, "first", {
+      debounceMs: 5,
+      steeringMode: "all",
+    });
     expect(
-      queueActiveRunMessageForTest("session-1", "first", { debounceMs: 5, steeringMode: "all" }),
-    ).toBe(true);
-    expect(
-      queueActiveRunMessageForTest("session-1", "second", { debounceMs: 5, steeringMode: "all" }),
+      queueActiveRunMessageForTest(params.sessionId, "second", {
+        debounceMs: 5,
+        steeringMode: "all",
+      }),
     ).toBe(true);
 
     await vi.waitFor(
@@ -193,7 +211,7 @@ describe("runCodexAppServerAttempt steering", () => {
             },
           },
         ]),
-      { interval: 1 },
+      fastWait,
     );
 
     await completeTurn({ threadId: "thread-1", turnId: "turn-1" });
@@ -235,10 +253,7 @@ describe("runCodexAppServerAttempt steering", () => {
         }) as never,
     );
 
-    const params = createParams(
-      path.join(tempDir, "session.jsonl"),
-      path.join(tempDir, "workspace"),
-    );
+    const params = createSteeringParams("steering-request-input");
     params.onBlockReply = vi.fn();
     const run = runCodexAppServerAttempt(params);
     await vi.waitFor(
@@ -271,7 +286,7 @@ describe("runCodexAppServerAttempt steering", () => {
     });
 
     await vi.waitFor(() => expect(params.onBlockReply).toHaveBeenCalledTimes(1), fastWait);
-    expect(queueActiveRunMessageForTest("session-1", "2")).toBe(true);
+    await queueActiveRunMessageEventually(params.sessionId, "2");
     await expect(response).resolves.toEqual({
       answers: { mode: { answers: ["Deep"] } },
     });

extensions/codex/src/app-server/run-attempt.test.ts

@@ -3937,6 +3937,46 @@ describe("runCodexAppServerAttempt", () => {
     }
   });
 
+  it("does not install an active run handle when turn start resolves after abort", async () => {
+    let resolveTurnStart: ((value: ReturnType<typeof turnStartResult>) => void) | undefined;
+    const request = vi.fn(async (method: string) => {
+      if (method === "thread/start") {
+        return threadStartResult("thread-1");
+      }
+      if (method === "turn/start") {
+        return await new Promise<ReturnType<typeof turnStartResult>>((resolve) => {
+          resolveTurnStart = resolve;
+        });
+      }
+      return {};
+    });
+    setCodexAppServerClientFactoryForTest(
+      async () =>
+        ({
+          request,
+          addNotificationHandler: () => () => undefined,
+          addRequestHandler: () => () => undefined,
+        }) as never,
+    );
+    const abortController = new AbortController();
+    const params = createParams(
+      path.join(tempDir, "session.jsonl"),
+      path.join(tempDir, "workspace"),
+    );
+    params.abortSignal = abortController.signal;
+
+    const run = runCodexAppServerAttempt(params);
+    await vi.waitFor(
+      () => expect(request.mock.calls.map(([method]) => method)).toContain("turn/start"),
+      fastWait,
+    );
+    abortController.abort("test_abort");
+    resolveTurnStart?.(turnStartResult());
+
+    await expect(run).rejects.toThrow("test_abort");
+    expect(queueActiveRunMessageForTest("session-1", "after abort")).toBe(false);
+  });
+
   it("keeps extended history enabled when resuming a bound Codex thread", async () => {
     const sessionFile = path.join(tempDir, "session.jsonl");
     const workspaceDir = path.join(tempDir, "workspace");

extensions/codex/src/app-server/run-attempt.ts

@@ -1832,6 +1832,22 @@ export async function runCodexAppServerAttempt(
   });
 
   let turn: CodexTurnStartResponse | undefined;
+  const throwIfTurnStartAcceptedAfterAbort = () => {
+    if (!runAbortController.signal.aborted) {
+      return;
+    }
+    const reason = runAbortController.signal.reason;
+    if (reason instanceof Error) {
+      throw reason;
+    }
+    const error = new Error(
+      typeof reason === "string" && reason.length > 0
+        ? reason
+        : "codex app-server turn start aborted before acceptance",
+    );
+    error.name = "AbortError";
+    throw error;
+  };
   const startCodexTurn = async (): Promise<CodexTurnStartResponse> => {
     const turnStartParams = buildTurnStartParams(params, {
       threadId: thread.threadId,
@@ -1847,12 +1863,14 @@ export async function runCodexAppServerAttempt(
         workspaceBootstrapContext.heartbeatCollaborationInstructions,
     });
     codexModelCallDiagnostics.setRequestPayloadBytes(utf8JsonByteLength(turnStartParams));
-    return assertCodexTurnStartResponse(
+    const startedTurn = assertCodexTurnStartResponse(
       await client.request("turn/start", turnStartParams, {
         timeoutMs: params.timeoutMs,
         signal: runAbortController.signal,
       }),
     );
+    throwIfTurnStartAcceptedAfterAbort();
+    return startedTurn;
   };
   try {
     codexModelCallDiagnostics.emitStarted();
@@ -2123,7 +2141,7 @@ export async function runCodexAppServerAttempt(
     kind: "embedded" as const,
     queueMessage: async (text: string, optionsLocal?: CodexSteeringQueueOptions) =>
       activeSteeringQueue.queue(text, optionsLocal),
-    isStreaming: () => !completed,
+    isStreaming: () => !completed && !runAbortController.signal.aborted,
     isCompacting: () => projectorRef.current?.isCompacting() ?? false,
     sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,
     cancel: () => runAbortController.abort("cancelled"),

extensions/codex/src/app-server/version.ts

@@ -2,4 +2,4 @@ export const MIN_CODEX_APP_SERVER_VERSION = "0.125.0";
 export const MIN_CODEX_SANDBOX_EXEC_SERVER_APP_SERVER_VERSION = "0.132.0";
 export const MANAGED_CODEX_APP_SERVER_PACKAGE = "@openai/codex";
 // Keep this in sync with the Codex CLI live-test package pin.
-export const MANAGED_CODEX_APP_SERVER_PACKAGE_VERSION = "0.134.0";
+export const MANAGED_CODEX_APP_SERVER_PACKAGE_VERSION = "0.135.0";