From f1acc1d600937b8a33665c565dd5023f4b6bcc63 Mon Sep 17 00:00:00 2001 From: Bernhard Geisberger Date: Mon, 31 Aug 2026 16:30:31 +0000 Subject: [PATCH 1/2] fix: wait for MCP server startup before starting an agent turn Sessions created with MCP servers registered a pending startup entry and kicked off the status publisher as fire-and-forget, so newSession and loadSession returned before Codex had started the servers. A prompt sent right after could start a turn without the session's MCP tools. Retain the startup promise and await it in prompt() before dispatching a turn. Only prompts that actually run an agent turn wait: commands the adapter answers itself, and Codex requests that never run a turn, are classified by CodexCommands.startsAgentTurn() and dispatch immediately. The wait is bounded by MCP_STARTUP_PROMPT_TIMEOUT_MS (default 30s), because the startup result only settles once every requested server reports a status newer than the snapshot version, which Codex may never send. On timeout the turn starts without those tools and later prompts are not delayed again. Cancel during the wait is handled explicitly: no turn exists yet, so cancel() would otherwise have found nothing to interrupt. --- readme-dev.md | 1 + src/CodexAcpServer.ts | 114 ++++++-- src/CodexCommands.ts | 30 +++ .../CodexACPAgent/mcp-startup-gate.test.ts | 255 ++++++++++++++++++ 4 files changed, 385 insertions(+), 15 deletions(-) create mode 100644 src/__tests__/CodexACPAgent/mcp-startup-gate.test.ts diff --git a/readme-dev.md b/readme-dev.md index bc147807..0f0caa2c 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -12,6 +12,7 @@ Set `CODEX_PATH` to run a different Codex binary; versions other than the one sp - `INITIAL_AGENT_MODE` - initial mode id: `read-only`, `agent`, or `agent-full-access`. - `NO_BROWSER` - hide browser-based ChatGPT auth when set. - `APP_SERVER_LOGS` - directory for adapter logs. +- `MCP_STARTUP_PROMPT_TIMEOUT_MS` - how long a prompt waits for the session's MCP servers to finish starting before the turn is started without them (default `30000`). ### Quick start diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 3828e7c4..aeb53bdc 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -184,6 +184,8 @@ export interface SessionFailure { const CODEX_PROCESS_EXITED_ERROR_CODE = 1001; +const DEFAULT_MCP_STARTUP_PROMPT_TIMEOUT_MS = 30_000; + function clientSupportsTypedSessionFailures(capabilities: acp.ClientCapabilities | null): boolean { return clientSupportsAirCapability(capabilities, AIR_SESSION_FAILURE_KEY); } @@ -200,6 +202,8 @@ interface ActiveAuthState { interface PendingMcpStartupSession { requestedServers: Set; afterVersion: number; + settled: Promise; + gateExpired: boolean; } interface PendingTurnStart { @@ -213,6 +217,7 @@ interface ActivePrompt { cancelSignal: Promise; signal: AbortSignal; currentTurn: { threadId: string, turnId: string } | null; + awaitingMcpStartup: boolean; requestCancel: () => void; requestClose: () => void; complete: () => void; @@ -658,11 +663,7 @@ export class CodexAcpServer { const canPublishSessionUpdates = operation !== "fork"; if (canPublishSessionUpdates && requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { - this.pendingMcpStartupSessions.set(sessionId, { - requestedServers: new Set(getRequestedMcpServerNames(requestedMcpServers)), - afterVersion: mcpServerStartupVersion, - }); - this.publishMcpStartupStatusAsync(sessionId); + this.trackMcpServerStartup(sessionId, requestedMcpServers, mcpServerStartupVersion); } if (canPublishSessionUpdates) { @@ -1708,11 +1709,7 @@ export class CodexAcpServer { subscribed = false; if (requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { - this.pendingMcpStartupSessions.set(sessionId, { - requestedServers: new Set(getRequestedMcpServerNames(requestedMcpServers)), - afterVersion: mcpServerStartupVersion, - }); - this.publishMcpStartupStatusAsync(sessionId); + this.trackMcpServerStartup(sessionId, requestedMcpServers, mcpServerStartupVersion); } await this.publishAvailableCommands(sessionState, requestedSessionGeneration); @@ -2152,16 +2149,73 @@ export class CodexAcpServer { return []; } - private publishMcpStartupStatusAsync(sessionId: string): void { - void this.doPublishMcpStartupStatus(sessionId); + private trackMcpServerStartup( + sessionId: string, + requestedMcpServers: Array, + afterVersion: number, + ): void { + const pendingStartup: PendingMcpStartupSession = { + requestedServers: new Set(getRequestedMcpServerNames(requestedMcpServers)), + afterVersion: afterVersion, + settled: Promise.resolve(), + gateExpired: false, + }; + this.pendingMcpStartupSessions.set(sessionId, pendingStartup); + pendingStartup.settled = this.doPublishMcpStartupStatus(sessionId, pendingStartup); } - private async doPublishMcpStartupStatus(sessionId: string): Promise { + /** + * Waits until the session's MCP servers finished starting, so a turn never runs with tools + * Codex has not registered yet. Returns false when the prompt was cancelled while waiting. + */ + private async awaitSessionMcpStartup(sessionId: string, activePrompt: ActivePrompt): Promise { const pendingStartup = this.pendingMcpStartupSessions.get(sessionId); - if (!pendingStartup) { - return; + if (!pendingStartup || pendingStartup.gateExpired) { + return true; } + const requestedServers = Array.from(pendingStartup.requestedServers); + logger.log("Waiting for MCP server startup before starting a turn", {sessionId, servers: requestedServers}); + const timeoutMs = getMcpStartupPromptTimeoutMs(); + let timeoutHandle: ReturnType | undefined; + const expired = new Promise<"expired">((resolve) => { + timeoutHandle = setTimeout(() => resolve("expired"), timeoutMs); + timeoutHandle.unref?.(); + }); + activePrompt.awaitingMcpStartup = true; + try { + const outcome = await Promise.race([ + pendingStartup.settled.then(() => "started" as const), + activePrompt.cancelSignal.then(() => "cancelled" as const), + expired, + ]); + if (outcome === "cancelled") { + logger.log("Prompt cancelled while waiting for MCP server startup", {sessionId}); + return false; + } + if (outcome === "expired") { + // Codex may never report a post-startup status for a server. Do not block prompts + // forever on it: run the turn without those tools and stop waiting on later prompts. + pendingStartup.gateExpired = true; + logger.log("MCP server startup timed out, starting the turn without those tools", { + sessionId, + timeoutMs, + servers: requestedServers, + }); + return true; + } + logger.log("MCP server startup completed, starting the turn", {sessionId}); + return true; + } finally { + activePrompt.awaitingMcpStartup = false; + clearTimeout(timeoutHandle); + } + } + + private async doPublishMcpStartupStatus( + sessionId: string, + pendingStartup: PendingMcpStartupSession, + ): Promise { try { const mcpStartup = await this.runWithProcessCheck(() => this.codexAcpClient.awaitMcpServerStartup( @@ -2228,6 +2282,7 @@ export class CodexAcpServer { cancelSignal, signal: abortController.signal, currentTurn: null, + awaitingMcpStartup: false, requestCancel: () => { if (abortController.signal.aborted) { return; @@ -2523,6 +2578,15 @@ export class CodexAcpServer { return cancelledPromptResponse(); } + if (this.availableCommands.startsAgentTurn(params.prompt)) { + if (!await this.awaitSessionMcpStartup(params.sessionId, activePrompt)) { + return cancelledPromptResponse(); + } + if (this.sessionIsClosing(params.sessionId)) { + return cancelledPromptResponse(); + } + } + const commandPromise = this.availableCommands.tryHandleCommand(params.prompt, sessionState, { onTurnStartPending: () => { ensurePendingTurnStart(); @@ -3007,6 +3071,14 @@ export class CodexAcpServer { return; } + // No turn exists yet while the prompt waits for MCP startup, so there is nothing to interrupt. + const activePrompt = this.activePrompts.get(params.sessionId); + if (activePrompt?.awaitingMcpStartup) { + logger.log("Cancel requested while waiting for MCP server startup", {sessionId: params.sessionId}); + activePrompt.requestCancel(); + return; + } + // After turnInterrupt(), Codex will send turn/completed, which naturally completes awaitTurnCompleted(). await this.interruptSessionTurn(sessionState, "Cancel", false); } @@ -3098,3 +3170,15 @@ function historyUpdateContentKey(update: UpdateSessionEvent): string | null { function getRequestedMcpServerNames(mcpServers: Array): Array { return Array.from(new Set(mcpServers.map(server => sanitizeMcpServerName(server.name)))); } + +function getMcpStartupPromptTimeoutMs(): number { + const value = process.env["MCP_STARTUP_PROMPT_TIMEOUT_MS"]?.trim(); + if (!value) { + return DEFAULT_MCP_STARTUP_PROMPT_TIMEOUT_MS; + } + const configured = Number(value); + if (!Number.isFinite(configured) || configured < 0) { + return DEFAULT_MCP_STARTUP_PROMPT_TIMEOUT_MS; + } + return configured; +} diff --git a/src/CodexCommands.ts b/src/CodexCommands.ts index 002cfae1..c8a6a034 100644 --- a/src/CodexCommands.ts +++ b/src/CodexCommands.ts @@ -202,6 +202,36 @@ export class CodexCommands { }; } + /** + * Whether this prompt makes Codex run an agent turn that can call MCP tools. + * Keep the switch in sync with {@link tryHandleCommand}. + */ + startsAgentTurn(prompt: acp.ContentBlock[]): boolean { + const command = this.parseCommand(prompt); + if (command === null) return true; + if (command.name.startsWith("$")) return true; + + switch (command.name) { + case "plan": + case "status": + case "skills": + case "mcp": + case "rename": + case "logout": + case "compact": + return false; + case "review": + case "review-branch": + case "review-commit": + // "/goal pause" and "/goal clear" do not start a turn, but the other forms do. + case "goal": + return true; + default: + // Unrecognized commands are forwarded to Codex as raw prompts. + return true; + } + } + async tryHandleCommand( prompt: acp.ContentBlock[], sessionState: SessionState, diff --git a/src/__tests__/CodexACPAgent/mcp-startup-gate.test.ts b/src/__tests__/CodexACPAgent/mcp-startup-gate.test.ts new file mode 100644 index 00000000..1ccda06b --- /dev/null +++ b/src/__tests__/CodexACPAgent/mcp-startup-gate.test.ts @@ -0,0 +1,255 @@ +import {afterEach, describe, expect, it, vi} from "vitest"; +import { + createCodexMockTestFixture, + createTestModel, + mockPromptTurn, + type CodexMockTestFixture, +} from "../acp-test-utils"; +import type {CodexAcpServer} from "../../CodexAcpServer"; +import type {CodexAcpClient} from "../../CodexAcpClient"; +import type {McpStartupResult} from "../../CodexAppServerClient"; +import type {McpServer} from "@agentclientprotocol/sdk"; + +const sessionId = "session-id"; + +const mcpServer: McpServer = { + name: "test-mcp", + command: "npx", + args: ["test-mcp"], + env: [], +}; + +describe("MCP startup prompt gate", () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it("does not start a turn before the session MCP servers are ready", async () => { + const mcpStartup = deferred(); + const {fixture, codexAcpAgent, codexAcpClient} = await createSession({ + mcpServers: [mcpServer], + configure: ({codexAcpClient}) => { + vi.spyOn(codexAcpClient, "awaitMcpServerStartup").mockReturnValue(mcpStartup.promise); + }, + }); + await vi.waitFor(() => { + expect(codexAcpClient.awaitMcpServerStartup).toHaveBeenCalledWith(["test-mcp"], expect.any(Number)); + }); + + const turnStart = mockPromptTurn(fixture, sessionId); + const promptPromise = codexAcpAgent.prompt({ + sessionId, + prompt: [{type: "text", text: "use the mcp tool"}], + }); + await waitForMicrotasks(); + + expect(turnStart).not.toHaveBeenCalled(); + + mcpStartup.resolve({ready: ["test-mcp"], failed: [], cancelled: []}); + + await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"}); + expect(turnStart).toHaveBeenCalledTimes(1); + }); + + it("starts the turn immediately when the session has no MCP servers", async () => { + const {fixture, codexAcpAgent} = await createSession(); + + const turnStart = mockPromptTurn(fixture, sessionId); + const promptPromise = codexAcpAgent.prompt({ + sessionId, + prompt: [{type: "text", text: "no mcp needed"}], + }); + await waitForMicrotasks(); + + expect(turnStart).toHaveBeenCalledTimes(1); + await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"}); + }); + + it("waits only for the first prompt once startup completed", async () => { + const mcpStartup = deferred(); + const {fixture, codexAcpAgent, codexAcpClient} = await createSession({ + mcpServers: [mcpServer], + configure: ({codexAcpClient}) => { + vi.spyOn(codexAcpClient, "awaitMcpServerStartup").mockReturnValue(mcpStartup.promise); + }, + }); + await vi.waitFor(() => { + expect(codexAcpClient.awaitMcpServerStartup).toHaveBeenCalledTimes(1); + }); + mcpStartup.resolve({ready: ["test-mcp"], failed: [], cancelled: []}); + + const turnStart = mockPromptTurn(fixture, sessionId); + await codexAcpAgent.prompt({sessionId, prompt: [{type: "text", text: "first"}]}); + + const secondPrompt = codexAcpAgent.prompt({sessionId, prompt: [{type: "text", text: "second"}]}); + await waitForMicrotasks(); + + expect(turnStart).toHaveBeenCalledTimes(2); + await expect(secondPrompt).resolves.toMatchObject({stopReason: "end_turn"}); + }); + + it("does not delay commands the adapter answers itself", async () => { + const mcpStartup = deferred(); + const {fixture, codexAcpAgent, codexAcpClient} = await createSession({ + mcpServers: [mcpServer], + configure: ({codexAcpClient}) => { + vi.spyOn(codexAcpClient, "awaitMcpServerStartup").mockReturnValue(mcpStartup.promise); + }, + }); + await vi.waitFor(() => { + expect(codexAcpClient.awaitMcpServerStartup).toHaveBeenCalledTimes(1); + }); + + // Startup is still pending: a local command must answer without waiting for it. + await expect(codexAcpAgent.prompt({ + sessionId, + prompt: [{type: "text", text: "/status"}], + })).resolves.toMatchObject({stopReason: "end_turn"}); + expect(fixture.getAcpConnectionDump([])).toContain("Model:"); + + mcpStartup.resolve({ready: ["test-mcp"], failed: [], cancelled: []}); + }); + + it("waits for MCP server startup before a command starts a turn", async () => { + const mcpStartup = deferred(); + const {codexAcpAgent, codexAcpClient} = await createSession({ + mcpServers: [mcpServer], + configure: ({codexAcpClient}) => { + vi.spyOn(codexAcpClient, "awaitMcpServerStartup").mockReturnValue(mcpStartup.promise); + }, + }); + await vi.waitFor(() => { + expect(codexAcpClient.awaitMcpServerStartup).toHaveBeenCalledTimes(1); + }); + + const runReview = vi.spyOn(codexAcpClient, "runReview").mockResolvedValue({ + threadId: sessionId, + turn: { + id: "review-turn", + items: [], + itemsView: "notLoaded", + status: "completed", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }, + }); + const promptPromise = codexAcpAgent.prompt({ + sessionId, + prompt: [{type: "text", text: "/review"}], + }); + await waitForMicrotasks(); + + expect(runReview).not.toHaveBeenCalled(); + + mcpStartup.resolve({ready: ["test-mcp"], failed: [], cancelled: []}); + + await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"}); + expect(runReview).toHaveBeenCalledTimes(1); + }); + + it("cancels a prompt that is waiting for MCP server startup", async () => { + const mcpStartup = deferred(); + const {fixture, codexAcpAgent, codexAcpClient} = await createSession({ + mcpServers: [mcpServer], + configure: ({codexAcpClient}) => { + vi.spyOn(codexAcpClient, "awaitMcpServerStartup").mockReturnValue(mcpStartup.promise); + }, + }); + await vi.waitFor(() => { + expect(codexAcpClient.awaitMcpServerStartup).toHaveBeenCalledTimes(1); + }); + + const turnStart = mockPromptTurn(fixture, sessionId); + const promptPromise = codexAcpAgent.prompt({ + sessionId, + prompt: [{type: "text", text: "cancel me"}], + }); + await waitForMicrotasks(); + + await codexAcpAgent.cancel({sessionId}); + + await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"}); + expect(turnStart).not.toHaveBeenCalled(); + + mcpStartup.resolve({ready: ["test-mcp"], failed: [], cancelled: []}); + }); + + it("starts the turn when MCP startup never reports back", async () => { + vi.stubEnv("MCP_STARTUP_PROMPT_TIMEOUT_MS", "10"); + const neverSettles = new Promise(() => {}); + const {fixture, codexAcpAgent, codexAcpClient} = await createSession({ + mcpServers: [mcpServer], + configure: ({codexAcpClient}) => { + vi.spyOn(codexAcpClient, "awaitMcpServerStartup").mockReturnValue(neverSettles); + }, + }); + await vi.waitFor(() => { + expect(codexAcpClient.awaitMcpServerStartup).toHaveBeenCalledTimes(1); + }); + + const turnStart = mockPromptTurn(fixture, sessionId); + await expect(codexAcpAgent.prompt({ + sessionId, + prompt: [{type: "text", text: "first"}], + })).resolves.toMatchObject({stopReason: "end_turn"}); + + // The expired gate must not delay any later prompt again. + vi.stubEnv("MCP_STARTUP_PROMPT_TIMEOUT_MS", "100000"); + const secondPrompt = codexAcpAgent.prompt({sessionId, prompt: [{type: "text", text: "second"}]}); + await waitForMicrotasks(); + + expect(turnStart).toHaveBeenCalledTimes(2); + await expect(secondPrompt).resolves.toMatchObject({stopReason: "end_turn"}); + }); +}); + +async function createSession(options: { + mcpServers?: McpServer[], + configure?: (params: { + fixture: CodexMockTestFixture, + codexAcpAgent: CodexAcpServer, + codexAcpClient: CodexAcpClient, + }) => void, +} = {}): Promise<{ + fixture: CodexMockTestFixture, + codexAcpAgent: CodexAcpServer, + codexAcpClient: CodexAcpClient, +}> { + const fixture = createCodexMockTestFixture(); + const codexAcpAgent = fixture.getCodexAcpAgent(); + const codexAcpClient = fixture.getCodexAcpClient(); + + vi.spyOn(codexAcpClient, "authRequired").mockResolvedValue(false); + vi.spyOn(codexAcpClient, "getAccount").mockResolvedValue({account: null, requiresOpenaiAuth: false}); + vi.spyOn(codexAcpClient, "listSkills").mockResolvedValue({data: []}); + vi.spyOn(codexAcpClient, "newSession").mockResolvedValue({ + sessionId, + currentModelId: "model-id[medium]", + models: [createTestModel()], + collaborationMode: "default", + currentServiceTier: null, + additionalDirectories: [], + }); + + options.configure?.({fixture, codexAcpAgent, codexAcpClient}); + + await codexAcpAgent.newSession({cwd: "/test/cwd", mcpServers: options.mcpServers ?? []}); + fixture.clearCodexConnectionDump(); + fixture.clearAcpConnectionDump(); + + return {fixture, codexAcpAgent, codexAcpClient}; +} + +function deferred(): {promise: Promise, resolve: (value: T) => void} { + let resolve: (value: T) => void = () => {}; + const promise = new Promise((innerResolve) => { + resolve = innerResolve; + }); + return {promise, resolve}; +} + +async function waitForMicrotasks(): Promise { + await new Promise(resolve => setTimeout(resolve, 10)); +} From 0b0ed2867417ee21cb692c4c46d867770da0a06e Mon Sep 17 00:00:00 2001 From: Bernhard Geisberger <41961259+bgeisberger@users.noreply.github.com> Date: Mon, 7 Sep 2026 13:58:19 +0200 Subject: [PATCH 2/2] Refactor 'goal' command to validate arguments Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/CodexCommands.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/CodexCommands.ts b/src/CodexCommands.ts index c8a6a034..d3f0dc62 100644 --- a/src/CodexCommands.ts +++ b/src/CodexCommands.ts @@ -223,9 +223,12 @@ export class CodexCommands { case "review": case "review-branch": case "review-commit": - // "/goal pause" and "/goal clear" do not start a turn, but the other forms do. - case "goal": return true; + // "/goal pause", "/goal clear", and "/goal" (usage) do not start a turn, but the other forms do. + case "goal": { + const arg = command.rest.trim().toLowerCase(); + return !(arg.length === 0 || arg === "pause" || arg === "clear" || arg.length > 4000); + } default: // Unrecognized commands are forwarded to Codex as raw prompts. return true;