From 0c358d3ec79f0769cab7852378620e0288d18467 Mon Sep 17 00:00:00 2001 From: "Nikolai.Sviridov" Date: Wed, 8 Apr 2026 14:58:04 +0200 Subject: [PATCH] 0.118.0 rebase, update logic --- src/CodexAcpClient.ts | 18 ++ src/CodexAcpServer.ts | 110 ++++++++++- src/CodexAppServerClient.ts | 36 +++- src/CodexEventHandler.ts | 35 ++++ .../CodexACPAgent/CodexAcpClient.test.ts | 177 ++++++++++++++++++ 5 files changed, 373 insertions(+), 3 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 47d76716..99233106 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -7,6 +7,7 @@ import open from "open"; import type {Disposable} from "vscode-jsonrpc"; import type { ClientInfo, + McpStartupCompleteEvent, ReasoningEffort, ServerNotification } from "./app-server"; @@ -21,6 +22,7 @@ import type { GetAccountResponse, ListMcpServerStatusParams, ListMcpServerStatusResponse, + McpServerStatusUpdatedNotification, Model, SkillsListParams, SkillsListResponse, @@ -275,6 +277,22 @@ export class CodexAcpClient { return startup.ready; } + async awaitMcpStartupResult(mcpStartupVersion: number): Promise { + return await this.codexClient.awaitMcpStartup(mcpStartupVersion); + } + + onMcpServerStatusUpdated(handler: (event: McpServerStatusUpdatedNotification) => void): void { + this.codexClient.onMcpServerStatusUpdated(handler); + } + + getMcpServerStatusVersion(): number { + return this.codexClient.getMcpServerStatusVersion(); + } + + getMcpServerStatusUpdates(afterVersion: number): Array { + return this.codexClient.getMcpServerStatusUpdates(afterVersion); + } + getMcpStartupCompleteVersion(): number { return this.codexClient.getMcpStartupCompleteVersion(); } diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 3b695fa1..0ca3120a 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -10,9 +10,18 @@ import {CodexApprovalHandler} from "./CodexApprovalHandler"; import {CodexAuthMethods, type CodexAuthRequest} from "./CodexAuthMethod"; import {CodexAcpClient, type SessionMetadata, type SessionMetadataWithThread} from "./CodexAcpClient"; import {ACPSessionConnection, type UpdateSessionEvent} from "./ACPSessionConnection"; -import type {Account, CollabAgentToolCallStatus, Model, Thread, ThreadItem, UserInput, ReasoningEffortOption} from "./app-server/v2"; +import type {McpStartupCompleteEvent, InputModality, ReasoningEffort} from "./app-server"; +import type { + Account, + CollabAgentToolCallStatus, + McpServerStatusUpdatedNotification, + Model, + Thread, + ThreadItem, + UserInput, + ReasoningEffortOption +} from "./app-server/v2"; import type {RateLimitsMap} from "./RateLimitsMap"; -import type {InputModality, ReasoningEffort} from "./app-server"; import {ModelId} from "./ModelId"; import {AgentMode} from "./AgentMode"; import type {TokenCount} from "./TokenCount"; @@ -43,6 +52,12 @@ export interface SessionState { sessionMcpServers?: Array; } +interface PendingMcpStartupSession { + requestedServers: Set; + terminalServers: Set; + publishedFailures: Set; +} + export class CodexAcpServer implements acp.Agent { private readonly codexAcpClient: CodexAcpClient; private readonly connection: acp.AgentSideConnection; @@ -51,6 +66,7 @@ export class CodexAcpServer implements acp.Agent { private readonly availableCommands: CodexCommands; private readonly sessions: Map; + private readonly pendingMcpStartupSessions: Map; constructor( connection: acp.AgentSideConnection, @@ -59,6 +75,7 @@ export class CodexAcpServer implements acp.Agent { getExitCode?: () => number | null, ) { this.sessions = new Map(); + this.pendingMcpStartupSessions = new Map(); this.connection = connection; this.codexAcpClient = codexAcpClient; this.defaultAuthRequest = defaultAuthRequest ?? null; @@ -68,6 +85,9 @@ export class CodexAcpServer implements acp.Agent { codexAcpClient, (operation) => this.runWithProcessCheck(operation) ); + this.codexAcpClient.onMcpServerStatusUpdated((event) => { + void this.handleMcpServerStatusUpdated(event); + }); } async initialize( @@ -128,6 +148,7 @@ export class CodexAcpServer implements acp.Agent { async getOrCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, SessionModelState, SessionModeState]> { await this.checkAuthorization(); const mcpStartupVersion = this.codexAcpClient.getMcpStartupCompleteVersion(); + const mcpStatusVersion = this.codexAcpClient.getMcpServerStatusVersion(); let sessionMetadata: SessionMetadata; if ("sessionId" in request) { @@ -159,6 +180,17 @@ export class CodexAcpServer implements acp.Agent { } this.sessions.set(sessionId, sessionState); + const requestedMcpServers = request.mcpServers ?? []; + if (requestedMcpServers.length > 0) { + this.pendingMcpStartupSessions.set(sessionId, { + requestedServers: new Set(requestedMcpServers.map(server => server.name)), + terminalServers: new Set(), + publishedFailures: new Set(), + }); + await this.replayBufferedMcpStartupStatus(sessionId, mcpStatusVersion); + this.publishMcpStartupStatusAsync(sessionId, mcpStartupVersion); + } + this.publishAvailableCommandsAsync(sessionId); const sessionModelState: SessionModelState = this.createModelState(models, currentModelId); const sessionModeState: SessionModeState = sessionState.agentMode.toSessionModeState(); @@ -621,6 +653,80 @@ export class CodexAcpServer implements acp.Agent { return await this.runWithProcessCheck(() => this.codexAcpClient.awaitMcpStartup(mcpStartupVersion)); } + private publishMcpStartupStatusAsync(sessionId: string, mcpStartupVersion: number): void { + void (async () => { + try { + const mcpStartup = await this.runWithProcessCheck(() => this.codexAcpClient.awaitMcpStartupResult(mcpStartupVersion)); + const sessionState = this.sessions.get(sessionId); + if (sessionState) { + sessionState.sessionMcpServers = mcpStartup.ready; + } + await this.publishMcpStartupStatus(sessionId, mcpStartup); + this.pendingMcpStartupSessions.delete(sessionId); + } catch (err) { + logger.error(`Failed to publish MCP startup status for session ${sessionId}`, err); + } + })(); + } + + private async publishMcpStartupStatus(sessionId: string, mcpStartup: McpStartupCompleteEvent): Promise { + for (const update of CodexEventHandler.createMcpStartupUpdates(mcpStartup)) { + await this.connection.sessionUpdate({ + sessionId, + update, + }); + } + } + + private async handleMcpServerStatusUpdated(event: McpServerStatusUpdatedNotification): Promise { + for (const [sessionId, pending] of this.pendingMcpStartupSessions) { + if (!pending.requestedServers.has(event.name)) { + continue; + } + + if (event.status === "ready") { + pending.terminalServers.add(event.name); + const sessionState = this.sessions.get(sessionId); + if (sessionState && !(sessionState.sessionMcpServers ?? []).includes(event.name)) { + sessionState.sessionMcpServers = [...(sessionState.sessionMcpServers ?? []), event.name]; + } + } else if (event.status === "failed" || event.status === "cancelled") { + pending.terminalServers.add(event.name); + if (!pending.publishedFailures.has(event.name)) { + pending.publishedFailures.add(event.name); + const updates = CodexEventHandler.createMcpStartupUpdates({ + ready: [], + failed: event.status === "failed" ? [{ + server: event.name, + error: event.error ?? "Unknown MCP startup error", + }] : [], + cancelled: event.status === "cancelled" ? [event.name] : [], + }); + for (const update of updates) { + await this.connection.sessionUpdate({ + sessionId, + update, + }); + } + } + } + + if (pending.terminalServers.size >= pending.requestedServers.size) { + this.pendingMcpStartupSessions.delete(sessionId); + } + } + } + + private async replayBufferedMcpStartupStatus(sessionId: string, afterVersion: number): Promise { + const updates = this.codexAcpClient.getMcpServerStatusUpdates(afterVersion); + for (const update of updates) { + await this.handleMcpServerStatusUpdated(update); + if (!this.pendingMcpStartupSessions.has(sessionId)) { + return; + } + } + } + async prompt(params: acp.PromptRequest): Promise { logger.log("Prompt received", { sessionId: params.sessionId, diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index d9c9fda0..16c17ff2 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -11,6 +11,7 @@ import type { GetAccountParams, GetAccountResponse, LoginAccountParams, LoginAccountResponse, LogoutAccountResponse, ModelListParams, ModelListResponse, + McpServerStatusUpdatedNotification, ThreadStartParams, ThreadStartResponse, ThreadLoadedListParams, @@ -63,6 +64,12 @@ export class CodexAppServerClient { private mcpStartupCompleteVersion = 0; private lastMcpStartupComplete: McpStartupCompleteEvent | null = null; private readonly mcpStartupCompleteResolvers: Array> = []; + private mcpServerStatusVersion = 0; + private readonly mcpServerStatusUpdatedHandlers: Array<(event: McpServerStatusUpdatedNotification) => void> = []; + private readonly mcpServerStatusHistory: Array<{ + version: number; + event: McpServerStatusUpdatedNotification; + }> = []; constructor(connection: MessageConnection) { this.connection = connection; @@ -76,8 +83,10 @@ export class CodexAppServerClient { } return; } - const serverNotification = data as ServerNotification; + if (serverNotification.method === "mcpServer/startupStatus/updated") { + this.recordMcpServerStatusUpdated(serverNotification.params); + } this.notify(serverNotification); for (const callback of this.codexEventHandlers) { callback({ eventType: "notification", ...serverNotification }); @@ -166,6 +175,20 @@ export class CodexAppServerClient { ); } + onMcpServerStatusUpdated(handler: (event: McpServerStatusUpdatedNotification) => void): void { + this.mcpServerStatusUpdatedHandlers.push(handler); + } + + getMcpServerStatusVersion(): number { + return this.mcpServerStatusVersion; + } + + getMcpServerStatusUpdates(afterVersion: number): Array { + return this.mcpServerStatusHistory + .filter(entry => entry.version > afterVersion) + .map(entry => entry.event); + } + async accountRead(params: GetAccountParams): Promise { return await this.sendRequest({ method: "account/read", params: params }); } @@ -237,6 +260,17 @@ export class CodexAppServerClient { }); } + private recordMcpServerStatusUpdated(event: McpServerStatusUpdatedNotification): void { + this.mcpServerStatusVersion += 1; + this.mcpServerStatusHistory.push({ + version: this.mcpServerStatusVersion, + event, + }); + for (const handler of this.mcpServerStatusUpdatedHandlers) { + handler(event); + } + } + private async sendRequest(request: CodexRequest): Promise { for (const callback of this.codexEventHandlers) { callback({ eventType: "request", ...request}); diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 00e27310..762df553 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -20,6 +20,7 @@ import type { ThreadTokenUsageUpdatedNotification, TurnPlanUpdatedNotification } from "./app-server/v2"; +import type { McpStartupCompleteEvent } from "./app-server"; import {toTokenCount} from "./TokenCount"; import { createCommandExecutionUpdate, @@ -258,6 +259,40 @@ export class CodexEventHandler { } } + static createMcpStartupUpdates(event: McpStartupCompleteEvent): UpdateSessionEvent[] { + const failedUpdates = event.failed.map((server: McpStartupCompleteEvent["failed"][number]) => this.createMcpStartupToolCallUpdate( + server.server, + `[codex-acp forwarded startup error] MCP server \`${server.server}\` failed to start: ${server.error}` + )); + const cancelledUpdates = event.cancelled.map((server: McpStartupCompleteEvent["cancelled"][number]) => this.createMcpStartupToolCallUpdate( + server, + `[codex-acp forwarded startup error] MCP server \`${server}\` startup was cancelled.` + )); + + return [...failedUpdates, ...cancelledUpdates]; + } + + private static createMcpStartupToolCallUpdate(serverName: string, message: string): UpdateSessionEvent { + return { + sessionUpdate: "tool_call", + toolCallId: this.getMcpStartupToolCallId(serverName), + kind: "other", + title: `mcp__${serverName}__startup`, + status: "failed", + content: [{ + type: "content", + content: { + type: "text", + text: message, + }, + }], + }; + } + + private static getMcpStartupToolCallId(serverName: string): string { + return `mcp_startup.${encodeURIComponent(serverName)}`; + } + private completeCommandExecutionEvent(item: ThreadItem & { "type": "commandExecution" }): UpdateSessionEvent { return { sessionUpdate: "tool_call_update", diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 94883384..388cc1f1 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -277,6 +277,183 @@ describe('ACP server test', { timeout: 40_000 }, () => { expect(mcpServers).toEqual(["alpha", "beta"]); }); + it('forwards failed MCP startup as failed tool call updates after new session', async () => { + const mockFixture = createCodexMockTestFixture(); + const codexAcpAgent = mockFixture.getCodexAcpAgent(); + const codexAppServerClient = mockFixture.getCodexAppServerClient(); + + vi.spyOn(codexAcpAgent, "checkAuthorization").mockResolvedValue(undefined); + vi.spyOn(codexAppServerClient, "threadStart").mockResolvedValue({ + thread: { id: "thread-id" } as any, + model: "gpt-5", + reasoningEffort: "medium", + } as any); + vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ + data: [{ + id: "gpt-5", + name: "GPT-5", + inputModalities: ["text"], + supportedReasoningEfforts: [], + }], + hasMore: false, + } as any); + vi.spyOn(codexAppServerClient, "accountRead").mockResolvedValue({ + requiresOpenaiAuth: false, + account: null, + } as any); + vi.spyOn(codexAppServerClient, "listSkills").mockResolvedValue({ data: [] }); + const mcpServer = { + name: "broken-mcp", + command: "npx", + args: ["broken"], + env: [], + } as unknown as acp.McpServerStdio; + + const session = await codexAcpAgent.newSession({ + cwd: "/workspace", + mcpServers: [mcpServer] + }); + + mockFixture.sendServerNotification({ + method: "codex/event/mcp_startup_complete", + params: { + msg: { + type: "mcp_startup_complete", + ready: [], + failed: [{ + server: "broken-mcp", + error: "boom", + }], + cancelled: [], + } + } + }); + + await vi.waitFor(() => { + const dump = mockFixture.getAcpConnectionDump([]); + expect(dump).toContain('"sessionId": "thread-id"'); + expect(dump).toContain('"sessionUpdate": "tool_call"'); + expect(dump).toContain('"toolCallId": "mcp_startup.broken-mcp"'); + expect(dump).toContain('MCP server `broken-mcp` failed to start: boom'); + }); + + expect(session.sessionId).toBe("thread-id"); + }); + + it('forwards MCP startup failure from startupStatus notification after new session', async () => { + const mockFixture = createCodexMockTestFixture(); + const codexAcpAgent = mockFixture.getCodexAcpAgent(); + const codexAppServerClient = mockFixture.getCodexAppServerClient(); + + vi.spyOn(codexAcpAgent, "checkAuthorization").mockResolvedValue(undefined); + vi.spyOn(codexAppServerClient, "threadStart").mockResolvedValue({ + thread: { id: "thread-id" } as any, + model: "gpt-5", + reasoningEffort: "medium", + } as any); + vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ + data: [{ + id: "gpt-5", + name: "GPT-5", + inputModalities: ["text"], + supportedReasoningEfforts: [], + }], + hasMore: false, + } as any); + vi.spyOn(codexAppServerClient, "accountRead").mockResolvedValue({ + requiresOpenaiAuth: false, + account: null, + } as any); + vi.spyOn(codexAppServerClient, "listSkills").mockResolvedValue({ data: [] }); + const mcpServer = { + name: "broken-mcp", + command: "npx", + args: ["broken"], + env: [], + } as unknown as acp.McpServerStdio; + + await codexAcpAgent.newSession({ + cwd: "/workspace", + mcpServers: [mcpServer] + }); + + mockFixture.sendServerNotification({ + method: "mcpServer/startupStatus/updated", + params: { + name: "broken-mcp", + status: "failed", + error: "boom", + } + }); + + await vi.waitFor(() => { + const dump = mockFixture.getAcpConnectionDump([]); + expect(dump).toContain('"sessionId": "thread-id"'); + expect(dump).toContain('"sessionUpdate": "tool_call"'); + expect(dump).toContain('"toolCallId": "mcp_startup.broken-mcp"'); + expect(dump).toContain('MCP server `broken-mcp` failed to start: boom'); + }); + }); + + it('replays MCP startup failure that arrives before session state is registered', async () => { + const mockFixture = createCodexMockTestFixture(); + const codexAcpAgent = mockFixture.getCodexAcpAgent(); + const codexAcpClient = mockFixture.getCodexAcpClient(); + const codexAppServerClient = mockFixture.getCodexAppServerClient(); + + vi.spyOn(codexAcpAgent, "checkAuthorization").mockResolvedValue(undefined); + vi.spyOn(codexAppServerClient, "threadStart").mockImplementation(async () => { + mockFixture.sendServerNotification({ + method: "mcpServer/startupStatus/updated", + params: { + name: "broken-mcp", + status: "failed", + error: "boom", + } + }); + + return { + thread: { id: "thread-id" } as any, + model: "gpt-5", + reasoningEffort: "medium", + } as any; + }); + vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ + data: [{ + id: "gpt-5", + name: "GPT-5", + inputModalities: ["text"], + supportedReasoningEfforts: [], + }], + hasMore: false, + } as any); + vi.spyOn(codexAppServerClient, "accountRead").mockResolvedValue({ + requiresOpenaiAuth: false, + account: null, + } as any); + vi.spyOn(codexAppServerClient, "listSkills").mockResolvedValue({ data: [] }); + + const mcpServer = { + name: "broken-mcp", + command: "npx", + args: ["broken"], + env: [], + } as unknown as acp.McpServerStdio; + + await codexAcpAgent.newSession({ + cwd: "/workspace", + mcpServers: [mcpServer] + }); + + await vi.waitFor(() => { + const dump = mockFixture.getAcpConnectionDump([]); + expect(dump).toContain('"toolCallId": "mcp_startup.broken-mcp"'); + expect(dump).toContain('MCP server `broken-mcp` failed to start: boom'); + }); + + expect(codexAcpClient.getMcpServerStatusUpdates(0)).toHaveLength(1); + }); + it('prefetches session additional skill roots before turn start', async () => { const mockFixture = createCodexMockTestFixture(); const codexAcpAgent = mockFixture.getCodexAcpAgent();