diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts new file mode 100644 index 000000000..aa81e035b --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts @@ -0,0 +1,547 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { expect, test } from "vitest"; + +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import { AgentManager, type AgentManagerEvent } from "../agent-manager.js"; +import type { + AgentClient, + AgentLaunchContext, + AgentSession, + AgentSessionConfig, +} from "../agent-sdk-types.js"; +import { CodexAppServerAgentClient, CodexAppServerAgentSession } from "./codex-app-server-agent.js"; +import { + createFakeCodexAppServer, + type FakeCodexAppServer, +} from "./codex/test-utils/fake-app-server.js"; + +const logger = createTestLogger(); + +class ProcessExitCodexClient extends CodexAppServerAgentClient implements AgentClient { + constructor(private readonly appServers: FakeCodexAppServer[]) { + super(logger); + } + + override async isAvailable(): Promise { + return true; + } + + override async createSession( + config: AgentSessionConfig, + launchContext?: AgentLaunchContext, + ): Promise { + const session = new CodexAppServerAgentSession( + config, + null, + logger, + async () => { + const appServer = this.appServers.shift(); + if (!appServer) { + throw new Error("No fake Codex app-server available"); + } + return appServer.child; + }, + {}, + false, + false, + false, + launchContext?.agentId, + ); + await session.connect(); + return session; + } +} + +test("unexpected Codex app-server exit fails the active run and agent", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-exit-")); + const appServer = createFakeCodexAppServer(); + const manager = new AgentManager({ + clients: { codex: new ProcessExitCodexClient([appServer]) }, + logger, + }); + const events: AgentManagerEvent[] = []; + const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + undefined, + { workspaceId: undefined }, + ); + agentId = agent.id; + const run = manager.runAgent(agent.id, "keep working"); + const turnStart = await appServer.waitForTurnStart(); + appServer.startsTurn({ threadId: String(turnStart.threadId), turnId: "native-turn-1" }); + await manager.waitForAgentRunStart(agent.id); + + appServer.child.stderr.write("provider crashed"); + appServer.child.emit("exit", 17, null); + appServer.child.emit("exit", 17, null); + + await expect(run).rejects.toThrow( + "Codex app-server exited with code 17 and signal null\nprovider crashed", + ); + await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("error"); + expect(manager.getAgent(agent.id)?.lastError).toBe( + "Codex app-server exited with code 17 and signal null\nprovider crashed", + ); + expect(events).toContainEqual( + expect.objectContaining({ + type: "agent_stream", + agentId: agent.id, + event: expect.objectContaining({ + type: "turn_failed", + error: "Codex app-server exited with code 17 and signal null\nprovider crashed", + }), + }), + ); + expect( + events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"), + ).toHaveLength(1); + } finally { + if (agentId && manager.getAgent(agentId)) { + await manager.closeAgent(agentId); + } + unsubscribe(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("intentional agent close stays clean while a Codex turn is active", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-close-")); + const appServer = createFakeCodexAppServer(); + const manager = new AgentManager({ + clients: { codex: new ProcessExitCodexClient([appServer]) }, + logger, + }); + const events: AgentManagerEvent[] = []; + const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false }); + + try { + const agent = await manager.createAgent( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + undefined, + { workspaceId: undefined }, + ); + const run = manager.streamAgent(agent.id, "keep working"); + const drainRun = (async () => { + const runEvents = []; + for await (const event of run) { + runEvents.push(event); + } + return runEvents; + })(); + const turnStart = await appServer.waitForTurnStart(); + appServer.startsTurn({ threadId: String(turnStart.threadId), turnId: "native-turn-1" }); + await manager.waitForAgentRunStart(agent.id); + + await manager.closeAgent(agent.id); + + await expect(drainRun).resolves.toContainEqual( + expect.objectContaining({ type: "turn_canceled", reason: "agent closed" }), + ); + expect( + events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"), + ).toHaveLength(0); + expect(events).toContainEqual( + expect.objectContaining({ + type: "agent_state", + agent: expect.objectContaining({ id: agent.id, lifecycle: "closed" }), + }), + ); + } finally { + unsubscribe(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("idle Codex app-server exit reconnects without publishing a failed turn", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-idle-exit-")); + const exitedAppServer = createFakeCodexAppServer(); + const replacementAppServer = createFakeCodexAppServer(); + const manager = new AgentManager({ + clients: { + codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]), + }, + logger, + }); + const events: AgentManagerEvent[] = []; + const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + undefined, + { workspaceId: undefined }, + ); + agentId = agent.id; + + exitedAppServer.child.emit("exit", 17, null); + + await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + expect( + events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"), + ).toHaveLength(0); + + const run = manager.runAgent(agent.id, "continue after reconnect"); + const turnStart = await replacementAppServer.waitForTurnStart(); + replacementAppServer.startsTurn({ + threadId: String(turnStart.threadId), + turnId: "native-turn-2", + }); + replacementAppServer.completeTurn({ threadId: String(turnStart.threadId) }); + + await expect(run).resolves.toMatchObject({ sessionId: "thread-1" }); + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + } finally { + if (agentId && manager.getAgent(agentId)) { + await manager.closeAgent(agentId); + } + unsubscribe(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("failed reconnect preserves manager events for a later successful run", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-reconnect-failure-")); + const exitedAppServer = createFakeCodexAppServer(); + const failedReconnect = createFakeCodexAppServer({ + initialize: async () => { + throw new Error("reconnect failed"); + }, + }); + const successfulReconnect = createFakeCodexAppServer(); + const manager = new AgentManager({ + clients: { + codex: new ProcessExitCodexClient([exitedAppServer, failedReconnect, successfulReconnect]), + }, + logger, + }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + undefined, + { workspaceId: undefined }, + ); + agentId = agent.id; + exitedAppServer.child.emit("exit", 17, null); + + await expect(manager.runAgent(agent.id, "first reconnect")).rejects.toThrow("reconnect failed"); + + const run = manager.runAgent(agent.id, "second reconnect"); + const turnStart = await successfulReconnect.waitForTurnStart(); + successfulReconnect.startsTurn({ + threadId: String(turnStart.threadId), + turnId: "native-turn-3", + }); + successfulReconnect.completeTurn({ threadId: String(turnStart.threadId) }); + + await expect(run).resolves.toMatchObject({ sessionId: "thread-1" }); + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + } finally { + if (agentId && manager.getAgent(agentId)) { + await manager.closeAgent(agentId); + } + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("unexpected exit fails an autonomous Codex turn", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-autonomous-exit-")); + const appServer = createFakeCodexAppServer(); + const manager = new AgentManager({ + clients: { codex: new ProcessExitCodexClient([appServer]) }, + logger, + }); + const events: AgentManagerEvent[] = []; + const unsubscribe = manager.subscribe((event) => events.push(event), { replayState: false }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + undefined, + { workspaceId: undefined }, + ); + agentId = agent.id; + appServer.startsTurn({ threadId: "thread-1", turnId: "autonomous-turn" }); + await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("running"); + + appServer.child.stderr.write("autonomous provider crashed"); + appServer.child.emit("exit", 23, null); + + await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("error"); + expect(manager.getAgent(agent.id)?.lastError).toBe( + "Codex app-server exited with code 23 and signal null\nautonomous provider crashed", + ); + expect( + events.filter((event) => event.type === "agent_stream" && event.event.type === "turn_failed"), + ).toHaveLength(1); + } finally { + if (agentId && manager.getAgent(agentId)) { + await manager.closeAgent(agentId); + } + unsubscribe(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("provider permissions do not survive an unexpected exit and reconnect", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-permission-exit-")); + const exitedAppServer = createFakeCodexAppServer(); + const replacementAppServer = createFakeCodexAppServer(); + const manager = new AgentManager({ + clients: { + codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]), + }, + logger, + }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + undefined, + { workspaceId: undefined }, + ); + agentId = agent.id; + + const failedRun = manager.runAgent(agent.id, "run a command"); + const failedTurnStart = await exitedAppServer.waitForTurnStart(); + const failedThreadId = String(failedTurnStart.threadId); + exitedAppServer.startsTurn({ threadId: failedThreadId, turnId: "native-turn-1" }); + exitedAppServer.requestCommandApproval({ + itemId: "stale-command", + threadId: failedThreadId, + turnId: "native-turn-1", + command: "git status", + cwd: workdir, + reason: "confirm command", + }); + await expect + .poll(() => manager.getPendingPermissions(agent.id).map((request) => request.id)) + .toEqual(["permission-stale-command"]); + + exitedAppServer.child.emit("exit", 17, null); + + await expect(failedRun).rejects.toThrow("Codex app-server exited with code 17"); + expect(manager.getPendingPermissions(agent.id)).toEqual([]); + + const recoveredRun = manager.runAgent(agent.id, "continue after reconnect"); + const recoveredTurnStart = await replacementAppServer.waitForTurnStart(); + const recoveredThreadId = String(recoveredTurnStart.threadId); + replacementAppServer.startsTurn({ + threadId: recoveredThreadId, + turnId: "native-turn-2", + }); + replacementAppServer.requestCommandApproval({ + itemId: "new-command", + threadId: recoveredThreadId, + turnId: "native-turn-2", + command: "git diff", + cwd: workdir, + reason: "confirm command", + }); + await expect + .poll(() => manager.getPendingPermissions(agent.id).map((request) => request.id)) + .toEqual(["permission-new-command"]); + + await manager.respondToPermission(agent.id, "permission-new-command", { + behavior: "allow", + }); + + expect(manager.getPendingPermissions(agent.id)).toEqual([]); + replacementAppServer.completeTurn({ threadId: recoveredThreadId }); + await expect(recoveredRun).resolves.toMatchObject({ sessionId: "thread-1" }); + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + } finally { + if (agentId && manager.getAgent(agentId)) { + await manager.closeAgent(agentId); + } + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("idle provider exit preserves a synthetic plan approval across reconnect", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-plan-exit-")); + const exitedAppServer = createFakeCodexAppServer(); + const replacementAppServer = createFakeCodexAppServer(); + const manager = new AgentManager({ + clients: { + codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]), + }, + logger, + }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent( + { + provider: "codex", + cwd: workdir, + modeId: "auto", + model: "gpt-5.4", + featureValues: { plan_mode: true }, + }, + undefined, + { workspaceId: undefined }, + ); + agentId = agent.id; + + const planRun = manager.runAgent(agent.id, "make a plan"); + const planTurnStart = await exitedAppServer.waitForTurnStart(); + const planThreadId = String(planTurnStart.threadId); + exitedAppServer.startsTurn({ threadId: planThreadId, turnId: "plan-turn" }); + exitedAppServer.updatesPlan({ + threadId: planThreadId, + steps: ["Inspect the provider lifecycle", "Implement the fix"], + }); + exitedAppServer.completeTurn({ threadId: planThreadId }); + await expect(planRun).resolves.toMatchObject({ sessionId: "thread-1" }); + const [planApproval] = manager.getPendingPermissions(agent.id); + expect(planApproval).toMatchObject({ + name: "CodexPlanApproval", + kind: "plan", + }); + + exitedAppServer.child.emit("exit", 17, null); + + await expect.poll(() => manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + expect(manager.getPendingPermissions(agent.id)).toEqual([planApproval]); + const approval = await manager.respondToPermission(agent.id, planApproval!.id, { + behavior: "allow", + selectedActionId: "implement", + }); + expect(approval).toMatchObject({ + followUpPrompt: expect.stringContaining("Implement the fix"), + }); + + const implementationRun = manager.runAgent(agent.id, approval!.followUpPrompt!); + const implementationTurnStart = await replacementAppServer.waitForTurnStart(); + const implementationThreadId = String(implementationTurnStart.threadId); + replacementAppServer.startsTurn({ + threadId: implementationThreadId, + turnId: "implementation-turn", + }); + replacementAppServer.completeTurn({ threadId: implementationThreadId }); + await expect(implementationRun).resolves.toMatchObject({ sessionId: "thread-1" }); + expect(manager.getAgent(agent.id)?.lifecycle).toBe("idle"); + } finally { + if (agentId && manager.getAgent(agentId)) { + await manager.closeAgent(agentId); + } + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("concurrent session APIs share one reconnect after an idle provider exit", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-concurrent-reconnect-")); + const appServers = [ + createFakeCodexAppServer(), + createFakeCodexAppServer(), + createFakeCodexAppServer(), + ]; + let spawnCount = 0; + let releaseReconnect: (() => void) | undefined; + const reconnectGate = new Promise((resolve) => { + releaseReconnect = resolve; + }); + const session = new CodexAppServerAgentSession( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + null, + logger, + async () => { + const appServer = appServers[spawnCount]; + if (!appServer) { + throw new Error("No fake Codex app-server available"); + } + spawnCount += 1; + if (spawnCount > 1) { + await reconnectGate; + } + return appServer.child; + }, + ); + + try { + await session.connect(); + appServers[0]!.child.emit("exit", 17, null); + + const runtimeInfo = session.getRuntimeInfo(); + const turnStart = session.startTurn("continue after reconnect"); + 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); + } finally { + releaseReconnect?.(); + await session.close(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("session close disposes a provider that arrives from an in-flight reconnect", async () => { + const workdir = mkdtempSync(join(tmpdir(), "codex-process-close-reconnect-")); + const initialAppServer = createFakeCodexAppServer(); + const lateAppServer = createFakeCodexAppServer(); + let spawnCount = 0; + let markReconnectSpawned: (() => void) | undefined; + const reconnectSpawned = new Promise((resolve) => { + markReconnectSpawned = resolve; + }); + let releaseReconnect: (() => void) | undefined; + const reconnectGate = new Promise((resolve) => { + releaseReconnect = resolve; + }); + const session = new CodexAppServerAgentSession( + { provider: "codex", cwd: workdir, modeId: "auto", model: "gpt-5.4" }, + null, + logger, + async () => { + spawnCount += 1; + if (spawnCount === 1) { + return initialAppServer.child; + } + markReconnectSpawned?.(); + await reconnectGate; + return lateAppServer.child; + }, + ); + const events: AgentManagerEvent[] = []; + session.subscribe((event) => { + events.push({ type: "agent_stream", agentId: "test-agent", event }); + }); + const lateExit = new Promise((resolve) => { + lateAppServer.child.once("exit", () => resolve()); + }); + + try { + await session.connect(); + initialAppServer.child.emit("exit", 17, null); + + const turnStart = session.startTurn("continue after reconnect"); + await reconnectSpawned; + await session.close(); + releaseReconnect?.(); + + await expect(turnStart).rejects.toThrow("Codex app-server session is closed"); + await expect(lateExit).resolves.toBeUndefined(); + expect(events.filter((event) => event.event.type === "turn_failed")).toHaveLength(0); + + await expect(session.connect()).rejects.toThrow("Codex app-server session is closed"); + expect(spawnCount).toBe(2); + } finally { + releaseReconnect?.(); + await session.close(); + rmSync(workdir, { recursive: true, force: true }); + } +}); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index e17493f4e..2983ccee5 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -3164,6 +3164,8 @@ export class CodexAppServerAgentSession implements AgentSession { private unpairedCompactionNotificationCompletions = 0; private unpairedCompactionItemCompletions = 0; private connected = false; + private connectionPromise: Promise | null = null; + private closed = false; private collaborationModes: Array<{ name: string; mode?: string | null; @@ -3229,15 +3231,49 @@ export class CodexAppServerAgentSession implements AgentSession { } async connect(): Promise { + if (this.closed) { + throw this.createClosedError(); + } if (this.connected) return; + if (this.connectionPromise) { + await this.connectionPromise; + if (this.closed) { + throw this.createClosedError(); + } + return; + } + + const connectionPromise = this.establishConnection(); + this.connectionPromise = connectionPromise; + try { + await connectionPromise; + } finally { + if (this.connectionPromise === connectionPromise) { + this.connectionPromise = null; + } + } + if (this.closed) { + throw this.createClosedError(); + } + } + + private async establishConnection(): Promise { const child = await this.spawnAppServer(); - this.client = new CodexAppServerClient(child, this.logger, () => this.traceContext()); - this.client.setNotificationHandler((method, params) => this.handleNotification(method, params)); + const client = new CodexAppServerClient(child, this.logger, () => this.traceContext()); + if (this.closed) { + await client.dispose(); + throw this.createClosedError(); + } + this.client = client; + client.setUnexpectedTerminationHandler((error) => { + this.handleUnexpectedTermination(error); + }); + client.setNotificationHandler((method, params) => this.handleNotification(method, params)); this.registerRequestHandlers(); try { - await this.client.request("initialize", buildCodexAppServerInitializeParams()); - this.client.notify("initialized", {}); + await client.request("initialize", buildCodexAppServerInitializeParams()); + client.notify("initialized", {}); await this.loadCollaborationModes(); await this.loadSkills(); @@ -3249,20 +3285,31 @@ export class CodexAppServerAgentSession implements AgentSession { await this.loadPersistedHistory(); } + if (this.closed) { + throw this.createClosedError(); + } this.connected = true; } catch (error) { try { - await this.close(); - } catch (closeError) { + if (this.client === client) { + await this.disposeClient(); + } else { + await client.dispose(); + } + } catch (disposeError) { this.logger.warn( - { err: closeError, connectError: error }, - "Failed to close Codex app-server after connection failure", + { err: disposeError, connectError: error }, + "Failed to dispose Codex app-server client after connection failure", ); } throw error; } } + private createClosedError(): Error { + return new Error("Codex app-server session is closed"); + } + private traceContext(): CodexAppServerTraceContext { return { agentId: this.agentId, @@ -3271,6 +3318,24 @@ export class CodexAppServerAgentSession implements AgentSession { }; } + private handleUnexpectedTermination(error: Error): void { + this.connected = false; + const hasActiveRootTurn = this.activeForegroundTurnId !== null || this.currentTurnId !== null; + this.clearPendingPermissions({ preservePlanApprovals: !hasActiveRootTurn }); + if (hasActiveRootTurn) { + this.emitEvent({ + type: "turn_failed", + provider: CODEX_PROVIDER, + error: error.message, + }); + } + this.activeForegroundTurnId = null; + this.activeClientMessageId = null; + this.currentTurnId = null; + this.pendingForegroundTurnIdentification?.resolve(null); + this.pendingForegroundTurnIdentification = null; + } + private async loadCollaborationModes(): Promise { if (!this.client) return; try { @@ -4308,25 +4373,39 @@ export class CodexAppServerAgentSession implements AgentSession { } async close(): Promise { - for (const pending of this.pendingPermissionHandlers.values()) { - pending.resolve({ decision: "cancel" }); - } - this.pendingPermissionHandlers.clear(); - this.pendingPermissions.clear(); - this.resolvedPermissionRequests.clear(); + this.closed = true; + this.clearPendingPermissions(); this.pendingSubAgentNotificationsByThreadId.clear(); this.subscribers.clear(); this.activeForegroundTurnId = null; this.activeClientMessageId = null; this.pendingForegroundTurnIdentification?.resolve(null); this.pendingForegroundTurnIdentification = null; - if (this.client) { - await this.client.dispose(); + await this.disposeClient(); + this.currentThreadId = null; + } + + private clearPendingPermissions(options?: { preservePlanApprovals?: boolean }): void { + for (const [requestId, pending] of this.pendingPermissionHandlers) { + if (options?.preservePlanApprovals && pending.kind === "plan") { + continue; + } + pending.resolve({ decision: "cancel" }); + this.pendingPermissionHandlers.delete(requestId); + this.pendingPermissions.delete(requestId); } + this.mcpElicitationPermissionIds.clear(); + this.resolvedPermissionRequests.clear(); + } + + private async disposeClient(): Promise { + const client = this.client; this.client = null; this.connected = false; - this.currentThreadId = null; this.currentTurnId = null; + if (client) { + await client.dispose(); + } } async listCommands(): Promise { diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.ts index 81a58bb9b..b2be8c8eb 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.ts @@ -35,6 +35,7 @@ interface PendingRequest { type RequestHandler = (params: unknown, requestId: number) => unknown; type NotificationHandler = (method: string, params: unknown) => void; +type UnexpectedTerminationHandler = (error: Error) => void; export interface CodexThreadForkParams { threadId: string; @@ -159,6 +160,7 @@ export class CodexAppServerClient { private readonly pending = new Map(); private readonly requestHandlers = new Map(); private notificationHandler: NotificationHandler | null = null; + private unexpectedTerminationHandler: UnexpectedTerminationHandler | null = null; private nextId = 1; private disposed = false; private stderrBuffer = ""; @@ -184,12 +186,7 @@ export class CodexAppServerClient { child.on("error", (err) => { this.logger.error({ err }, "Codex app-server child process error"); - for (const pending of this.pending.values()) { - clearTimeout(pending.timer); - pending.reject(err); - } - this.pending.clear(); - this.disposed = true; + this.handleUnexpectedTermination(err); }); child.on("exit", (code, signal) => { @@ -198,15 +195,14 @@ export class CodexAppServerClient { ? "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()); - for (const pending of this.pending.values()) { - clearTimeout(pending.timer); - pending.reject(error); - } - this.pending.clear(); - this.disposed = true; + this.handleUnexpectedTermination(error); }); } + setUnexpectedTerminationHandler(handler: UnexpectedTerminationHandler): void { + this.unexpectedTerminationHandler = handler; + } + setNotificationHandler(handler: NotificationHandler): void { this.notificationHandler = handler; } @@ -251,6 +247,7 @@ export class CodexAppServerClient { async dispose(): Promise { if (this.disposed) return; this.disposed = true; + this.unexpectedTerminationHandler = null; this.rl.close(); try { this.child.stdin.end(); @@ -275,6 +272,29 @@ export class CodexAppServerClient { } } + private handleUnexpectedTermination(error: Error): void { + if (this.disposed) { + return; + } + this.disposed = true; + this.rl.close(); + for (const pending of this.pending.values()) { + clearTimeout(pending.timer); + pending.reject(error); + } + this.pending.clear(); + const handler = this.unexpectedTerminationHandler; + this.unexpectedTerminationHandler = null; + if (!handler) { + return; + } + try { + handler(error); + } catch (handlerError) { + this.logger.warn({ err: handlerError }, "Codex app-server termination handler threw"); + } + } + private writeJsonRpcResponse(response: JsonRpcResponse): void { if (this.disposed || this.child.stdin.destroyed || !this.child.stdin.writable) { return; diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index e9d6a10bd..dc04a31f0 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -51,6 +51,7 @@ export interface FakeCodexAppServer { waitForTurnStart(): Promise; nextResponse(): Promise; startsTurn(params: { threadId: string; turnId?: string }): void; + updatesPlan(params: { threadId: string; steps: string[] }): void; completeTurn(params?: { threadId?: string }): void; startsSubAgent(params: { callId: string; @@ -319,6 +320,17 @@ export function createFakeCodexAppServer( })}\n`, ); }, + updatesPlan(params) { + child.stdout.write( + `${JSON.stringify({ + method: "turn/plan/updated", + params: { + threadId: params.threadId, + plan: params.steps.map((step) => ({ step, status: "pending" })), + }, + })}\n`, + ); + }, completeTurn(params = {}) { child.stdout.write( `${JSON.stringify({