From ca3f247eb2150f628f023e17c41d1307d00a84c3 Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Wed, 8 Jul 2026 12:26:33 +0530 Subject: [PATCH] fix(codex): fan out app-server liveness retirement --- .../app-server/attempt-client-cleanup.test.ts | 24 ++++++ .../src/app-server/attempt-client-cleanup.ts | 6 +- .../app-server/attempt-turn-watches.test.ts | 74 +++++++++++++++++++ .../src/app-server/attempt-turn-watches.ts | 13 ++-- .../src/app-server/shared-client.test.ts | 72 +++++++++++++++--- .../codex/src/app-server/shared-client.ts | 30 +++++++- 6 files changed, 199 insertions(+), 20 deletions(-) diff --git a/extensions/codex/src/app-server/attempt-client-cleanup.test.ts b/extensions/codex/src/app-server/attempt-client-cleanup.test.ts index 9ffa38717e6d..65bffe676702 100644 --- a/extensions/codex/src/app-server/attempt-client-cleanup.test.ts +++ b/extensions/codex/src/app-server/attempt-client-cleanup.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from "vitest"; import { interruptCodexTurnBestEffort, + retireCodexAppServerClientAfterTimedOutTurn, unsubscribeCodexThreadBestEffort, } from "./attempt-client-cleanup.js"; @@ -40,4 +41,27 @@ describe("Codex app-server attempt client cleanup", () => { { timeoutMs: 123 }, ); }); + + it("closes only the isolated client after timed-out turn cleanup", async () => { + const request = vi.fn(async () => ({})); + const close = vi.fn(); + + await retireCodexAppServerClientAfterTimedOutTurn({ request, close } as never, { + threadId: "thread-1", + turnId: "turn-1", + reason: "turn_terminal_idle_timeout", + }); + + expect(request).toHaveBeenCalledWith( + "turn/interrupt", + { threadId: "thread-1", turnId: "turn-1" }, + { timeoutMs: 5_000 }, + ); + expect(request).toHaveBeenCalledWith( + "thread/unsubscribe", + { threadId: "thread-1" }, + { timeoutMs: 5_000 }, + ); + expect(close).toHaveBeenCalledTimes(1); + }); }); diff --git a/extensions/codex/src/app-server/attempt-client-cleanup.ts b/extensions/codex/src/app-server/attempt-client-cleanup.ts index 765e9cf78904..1a26bd1ccff8 100644 --- a/extensions/codex/src/app-server/attempt-client-cleanup.ts +++ b/extensions/codex/src/app-server/attempt-client-cleanup.ts @@ -146,7 +146,11 @@ export async function retireCodexAppServerClientAfterTimedOutTurn( reason: string; }, ): Promise { - const retiredSharedClient = retireSharedCodexAppServerClientIfCurrent(client); + // A timed-out turn marks the physical client suspect: fail co-leased + // attempts into their retry path instead of leaving them wedged on it. + const retiredSharedClient = retireSharedCodexAppServerClientIfCurrent(client, { + failActiveLeases: true, + }); const detachedSharedClient = Boolean(retiredSharedClient); interruptCodexTurnBestEffort(client, { threadId: params.threadId, diff --git a/extensions/codex/src/app-server/attempt-turn-watches.test.ts b/extensions/codex/src/app-server/attempt-turn-watches.test.ts index 9545b461cb14..b7aa6f2f9459 100644 --- a/extensions/codex/src/app-server/attempt-turn-watches.test.ts +++ b/extensions/codex/src/app-server/attempt-turn-watches.test.ts @@ -171,6 +171,80 @@ describe("Codex app-server attempt turn watches", () => { expect(harness.abortController.signal.aborted).toBe(false); }); + it("fires terminal idle timeout while an app-server request is in flight", () => { + const harness = createController(); + harness.activeRequests = 1; + + harness.controller.armTerminalIdleWatch(); + vi.advanceTimersByTime(10); + + expect(harness.timeouts).toMatchObject([ + { + kind: "terminal", + idleMs: 10, + timeoutMs: 10, + lastActivityReason: "startup", + }, + ]); + expect(harness.abortController.signal.reason).toBe("turn_terminal_idle_timeout"); + }); + + it("keeps terminal idle alive when notifications arrive during an in-flight request", () => { + const harness = createController(); + harness.activeRequests = 1; + + harness.controller.armTerminalIdleWatch(); + vi.advanceTimersByTime(9); + harness.controller.noteNotificationReceived("item/commandExecution/outputDelta"); + vi.advanceTimersByTime(9); + + expect(harness.timeouts).toEqual([]); + expect(harness.abortController.signal.aborted).toBe(false); + + vi.advanceTimersByTime(1); + expect(harness.timeouts).toMatchObject([ + { + kind: "terminal", + idleMs: 10, + timeoutMs: 10, + lastActivityReason: "notification:item/commandExecution/outputDelta", + }, + ]); + }); + + it("keeps completion idle gated while a request is in flight", () => { + const harness = createController(); + harness.activeRequests = 1; + + harness.controller.touchActivity("turn:start", { arm: true }); + vi.advanceTimersByTime(10); + + expect(harness.timeouts).toEqual([]); + expect(harness.abortController.signal.aborted).toBe(false); + + harness.activeRequests = 0; + harness.controller.scheduleProgressWatches(); + vi.advanceTimersByTime(1); + + expect(harness.timeouts).toMatchObject([{ kind: "completion" }]); + }); + + it("keeps assistant-completion gated while a request is in flight", () => { + const harness = createController(); + harness.activeRequests = 1; + + harness.controller.armAssistantCompletionIdleWatch(); + vi.advanceTimersByTime(10); + + expect(harness.completed).toBe(false); + expect(harness.abortController.signal.aborted).toBe(false); + + harness.activeRequests = 0; + vi.advanceTimersByTime(1); + + expect(harness.completed).toBe(true); + }); + it("waits for active completion blocker items before firing completion idle timeout", () => { const harness = createController(); harness.activeCompletionBlockers = 1; diff --git a/extensions/codex/src/app-server/attempt-turn-watches.ts b/extensions/codex/src/app-server/attempt-turn-watches.ts index f56327cc9a28..7bfbc0eb542b 100644 --- a/extensions/codex/src/app-server/attempt-turn-watches.ts +++ b/extensions/codex/src/app-server/attempt-turn-watches.ts @@ -166,12 +166,7 @@ export function createCodexAttemptTurnWatchController(params: { function scheduleTerminalIdleWatch() { clearTerminalIdleTimer(); - if ( - params.isCompleted() || - params.signal.aborted || - !terminalIdleWatchArmed || - params.getActiveAppServerTurnRequests() > 0 - ) { + if (params.isCompleted() || params.signal.aborted || !terminalIdleWatchArmed) { return; } const elapsedMs = Math.max(0, Date.now() - completionLastActivityAt); @@ -371,12 +366,14 @@ export function createCodexAttemptTurnWatchController(params: { } function fireTerminalIdleTimeout() { + // This is the physical-client liveness backstop. An in-flight request can + // be silent forever if the client is wedged, so only real notifications + // reset the terminal clock. if ( params.isCompleted() || params.isTerminalTurnNotificationQueued() || params.signal.aborted || - !terminalIdleWatchArmed || - params.getActiveAppServerTurnRequests() > 0 + !terminalIdleWatchArmed ) { return; } diff --git a/extensions/codex/src/app-server/shared-client.test.ts b/extensions/codex/src/app-server/shared-client.test.ts index e4c68ae531c9..14639f3c2e91 100644 --- a/extensions/codex/src/app-server/shared-client.test.ts +++ b/extensions/codex/src/app-server/shared-client.test.ts @@ -699,7 +699,7 @@ describe("shared Codex app-server client", () => { expect(second.process.stdin.destroyed).toBe(true); }); - it("closes a retired shared app-server after all active leases release", async () => { + it("closes a retired shared app-server and forces active leases onto the retryable close path", async () => { const first = createClientHarness(); const second = createClientHarness(); vi.spyOn(CodexAppServerClient, "start") @@ -715,11 +715,15 @@ describe("shared Codex app-server client", () => { const releaseSecond = retainSharedCodexAppServerClientIfCurrent(first.client); expect(releaseFirst).toBeTypeOf("function"); expect(releaseSecond).toBeTypeOf("function"); - expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ + const activeRequest = first.client.request("test/pending", {}); + expect( + retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), + ).toEqual({ activeLeases: 2, - closed: false, + closed: true, }); - expect(first.process.stdin.destroyed).toBe(false); + expect(first.process.stdin.destroyed).toBe(true); + await expect(activeRequest).rejects.toThrow("codex app-server client is closed"); const secondList = listCodexAppServerModels({ timeoutMs: 1000 }); await sendInitializeResult(second, "openclaw/0.143.0 (macOS; test)"); @@ -727,7 +731,6 @@ describe("shared Codex app-server client", () => { await expect(secondList).resolves.toEqual({ models: [] }); releaseFirst?.(); - expect(first.process.stdin.destroyed).toBe(false); releaseSecond?.(); expect(first.process.stdin.destroyed).toBe(true); expect(second.process.kill).not.toHaveBeenCalled(); @@ -748,21 +751,70 @@ describe("shared Codex app-server client", () => { await expect(firstLease).resolves.toBe(first.client); await expect(secondLease).resolves.toBe(first.client); - expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ + expect( + retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), + ).toEqual({ + activeLeases: 2, + closed: true, + }); + expect( + retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), + ).toEqual({ activeLeases: 2, closed: false, }); + expect(first.process.stdin.destroyed).toBe(true); + + expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); + expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); + expect(first.process.stdin.destroyed).toBe(true); + expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(false); + }); + + it("rejects pending acquires during shared-client retirement", async () => { + const first = createClientHarness(); + const second = createClientHarness(); + vi.spyOn(CodexAppServerClient, "start") + .mockReturnValueOnce(first.client) + .mockReturnValueOnce(second.client); + + const firstLease = getLeasedSharedCodexAppServerClient(); + const pendingLease = getLeasedSharedCodexAppServerClient(); + await vi.waitFor(() => expect(first.writes.length).toBeGreaterThanOrEqual(1)); + + expect( + retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), + ).toEqual({ + activeLeases: 0, + closed: true, + }); + await expect(firstLease).rejects.toThrow("codex app-server client is closed"); + await expect(pendingLease).rejects.toThrow("codex app-server client is closed"); + + const freshLease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 }); + await sendInitializeResult(second, "openclaw/0.142.0 (macOS; test)"); + await expect(freshLease).resolves.toBe(second.client); + expect(second.process.stdin.destroyed).toBe(false); + }); + + it("retires gracefully by default: leased clients close on release, not immediately", async () => { + const first = createClientHarness(); + vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(first.client); + + const lease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 }); + await sendInitializeResult(first, "openclaw/0.142.0 (macOS; test)"); + await expect(lease).resolves.toBe(first.client); + + // Routine cleanup (e.g. one-shot bundle-MCP) must not yank a healthy + // client from co-leased sessions; only suspect retirement does. expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ - activeLeases: 2, + activeLeases: 1, closed: false, }); expect(first.process.stdin.destroyed).toBe(false); - expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); - expect(first.process.stdin.destroyed).toBe(false); expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); expect(first.process.stdin.destroyed).toBe(true); - expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(false); }); it("waits only for the shared client that is still current", async () => { diff --git a/extensions/codex/src/app-server/shared-client.ts b/extensions/codex/src/app-server/shared-client.ts index 5186814cc34c..29d12a304b3b 100644 --- a/extensions/codex/src/app-server/shared-client.ts +++ b/extensions/codex/src/app-server/shared-client.ts @@ -26,6 +26,7 @@ type SharedCodexAppServerClientEntry = { activeLeases: number; pendingAcquires: number; closeWhenIdle: boolean; + closeError?: Error; }; type SharedCodexAppServerClientState = { @@ -212,6 +213,9 @@ async function acquireSharedCodexAppServerClient( options?.timeoutMs ?? 0, "codex app-server initialize timed out", ); + if (entry.closeError) { + throw entry.closeError; + } // Later leases of the same keyed client may carry fresher config; the // runtime install itself stays one-per-physical-client. ensureCodexAppServerClientRuntime(client, { @@ -408,9 +412,18 @@ export function retainSharedCodexAppServerClientIfCurrent( return undefined; } -/** Marks a matching shared client to close after active leases/acquires drain. */ +/** + * Retires a matching shared client. Default is graceful: detach from the map + * (future acquisitions get a fresh client) and close once leases drain. + * `failActiveLeases` is for suspect clients only (timed-out turns): it closes + * the physical connection immediately so co-leased attempts hit the normal + * client-closed retry path, and pending acquires reject instead of leasing + * the poisoned process. Routine cleanup must NOT use it — it would abort + * healthy sibling turns on a working client. + */ export function retireSharedCodexAppServerClientIfCurrent( client: CodexAppServerClient | undefined, + opts?: { failActiveLeases?: boolean }, ): { activeLeases: number; closed: boolean } | undefined { if (!client) { return undefined; @@ -420,6 +433,11 @@ export function retireSharedCodexAppServerClientIfCurrent( if (entry.client === client) { state.clients.delete(key); entry.closeWhenIdle = true; + if (opts?.failActiveLeases) { + entry.closeError = new Error("codex app-server client is closed"); + const closed = closeRetiredSharedClientEntry(entry); + return { activeLeases: entry.activeLeases, closed }; + } const closed = closeRetiredSharedClientEntryIfIdle(entry); return { activeLeases: entry.activeLeases, closed }; } @@ -556,6 +574,16 @@ function closeRetiredSharedClientEntryIfIdle(entry: SharedCodexAppServerClientEn return true; } +function closeRetiredSharedClientEntry(entry: SharedCodexAppServerClientEntry): boolean { + const client = entry.client; + if (!client) { + return false; + } + entry.client = undefined; + client.close(); + return true; +} + function closeSharedClientEntryIfUnclaimed( key: string, entry: SharedCodexAppServerClientEntry,