fix: reconcile transient MCP startup cancellations

src/__tests__/CodexACPAgent/mcp-startup-reconciliation.test.ts

@@ -0,0 +1,213 @@
+import {afterEach, describe, expect, it, vi} from "vitest";
+import type * as acp from "@agentclientprotocol/sdk/unstable";
+import type * as acpV2 from "@agentclientprotocol/sdk/experimental/v2/unstable";
+import type {ServerNotification} from "../../app-server";
+import type {McpServerStatus} from "../../app-server/v2";
+import {connectSession, type PromptSession, sessionId} from "./v2-prompt-harness";
+
+const versions = [1, 2] as const;
+const serverName = "everything";
+
+function server(runtimeStatus: McpServerStatus["runtimeStatus"], name = serverName): McpServerStatus {
+    return {
+        name,
+        runtimeStatus,
+        pluginId: null,
+        httpOrigin: null,
+        serverInfo: null,
+        serverCapabilities: null,
+        tools: {},
+        toolsError: null,
+        resources: [],
+        resourceTemplates: [],
+        authStatus: "unsupported",
+    };
+}
+
+async function connect(version: 1 | 2, status: (call: number) => McpServerStatus): Promise<{
+    client: PromptSession,
+    statusParams: unknown[],
+}> {
+    const statusParams: unknown[] = [];
+    const client = await connectSession(version, {
+        realMcpStartup: true,
+        ...(version === 2
+            ? {mcpServers: [{type: "stdio", name: serverName, command: "/usr/bin/everything"} as acpV2.McpServer]}
+            : {mcpServersV1: [{name: serverName, command: "/usr/bin/everything", args: [], env: []} as acp.McpServer]}),
+        codexResponses: {
+            "mcpServerStatus/list": async params => {
+                statusParams.push(params);
+                return {data: [status(statusParams.length)], nextCursor: null};
+            },
+        },
+    });
+    return {client, statusParams};
+}
+
+function emitCancelled(client: PromptSession, name = serverName): void {
+    const base = {threadId: sessionId, name, error: null, failureReason: null};
+    client.emit({method: "mcpServer/startupStatus/updated", params: {...base, status: "starting"}} as ServerNotification);
+    client.emit({method: "mcpServer/startupStatus/updated", params: {...base, status: "cancelled"}} as ServerNotification);
+}
+
+function startupFailures(client: PromptSession): Array<acpV2.SessionUpdate> {
+    return [...client.setupUpdates, ...client.transcript.flatMap(entry => "sessionUpdate" in entry ? [entry.sessionUpdate] : [])]
+        .filter(update => (update as {toolCallId?: string}).toolCallId?.startsWith("mcp_startup."));
+}
+
+function deferred<T>(): {promise: Promise<T>, resolve(value: T): void} {
+    let resolve!: (value: T) => void;
+    return {promise: new Promise<T>(done => resolve = done), resolve};
+}
+
+describe("MCP startup cancellation reconciliation", () => {
+    const clients: PromptSession[] = [];
+
+    afterEach(() => {
+        vi.useRealTimers();
+        clients.splice(0).forEach(client => client.connection.close());
+        vi.clearAllMocks();
+    });
+
+    it.each(versions)("ACP v%s suppresses a transient cancellation only after same-thread status is connected", async version => {
+        const {client, statusParams} = await connect(version, call => server(call === 1 ? "starting" : "connected"));
+        clients.push(client);
+        const requestsBeforeCancellation = client.codexRequests.length;
+
+        emitCancelled(client);
+
+        await vi.waitFor(() => expect(statusParams).toHaveLength(2));
+        expect(startupFailures(client)).toEqual([]);
+        expect(statusParams).toEqual([
+            expect.objectContaining({threadId: sessionId}),
+            expect.objectContaining({threadId: sessionId}),
+        ]);
+        expect(client.codexRequests.slice(requestsBeforeCancellation).map(request => request.method))
+            .toEqual(["mcpServerStatus/list", "mcpServerStatus/list"]);
+    });
+
+    it.each(versions)("ACP v%s preserves a genuine cancellation when status never connects", async version => {
+        const {client, statusParams} = await connect(version, () => server("cancelled"));
+        clients.push(client);
+        vi.useFakeTimers();
+
+        emitCancelled(client);
+        await vi.advanceTimersByTimeAsync(5_000);
+
+        expect(statusParams.length).toBeGreaterThan(1);
+        expect(startupFailures(client)).toHaveLength(1);
+        expect(startupFailures(client)[0]).toMatchObject({status: "failed"});
+        expect(statusParams.every(params => (params as {threadId?: string}).threadId === sessionId)).toBe(true);
+    });
+
+    it.each(versions)("ACP v%s rejects connected evidence from incomplete repeated-cursor pagination", async version => {
+        const statusParams: Array<{cursor?: string | null, threadId?: string}> = [];
+        const client = await connectSession(version, {
+            realMcpStartup: true,
+            ...(version === 2
+                ? {mcpServers: [{type: "stdio", name: serverName, command: "/usr/bin/everything"} as acpV2.McpServer]}
+                : {mcpServersV1: [{name: serverName, command: "/usr/bin/everything", args: [], env: []} as acp.McpServer]}),
+            codexResponses: {
+                "mcpServerStatus/list": async params => {
+                    const page = params as {cursor?: string | null, threadId?: string};
+                    statusParams.push(page);
+                    return page.cursor === null
+                        ? {data: [server("connected")], nextCursor: "repeated"}
+                        : {data: [], nextCursor: "repeated"};
+                },
+            },
+        });
+        clients.push(client);
+        vi.useFakeTimers();
+
+        emitCancelled(client);
+        await vi.advanceTimersByTimeAsync(5_000);
+
+        expect(startupFailures(client)).toHaveLength(1);
+        expect(startupFailures(client)[0]).toMatchObject({status: "failed"});
+        expect(statusParams.length).toBeGreaterThan(2);
+        expect(statusParams.every(params => params.threadId === sessionId)).toBe(true);
+        expect(statusParams.some(params => params.cursor === "repeated")).toBe(true);
+    });
+
+    it("publishes the cancellation at the deadline when the status request never settles", async () => {
+        const status = deferred<{data: McpServerStatus[], nextCursor: string | null}>();
+        const statusParams: unknown[] = [];
+        const client = await connectSession(2, {
+            realMcpStartup: true,
+            mcpServers: [{type: "stdio", name: serverName, command: "/usr/bin/everything"} as acpV2.McpServer],
+            codexResponses: {
+                "mcpServerStatus/list": async params => {
+                    statusParams.push(params);
+                    return await status.promise;
+                },
+            },
+        });
+        clients.push(client);
+        vi.useFakeTimers();
+
+        emitCancelled(client);
+        await vi.waitFor(() => expect(statusParams).toHaveLength(1));
+        expect(startupFailures(client)).toEqual([]);
+        await vi.advanceTimersByTimeAsync(5_000);
+
+        expect(startupFailures(client)).toHaveLength(1);
+        status.resolve({data: [server("connected")], nextCursor: "late-page"});
+        await vi.runAllTicks();
+        expect(statusParams).toHaveLength(1);
+        expect(startupFailures(client)).toHaveLength(1);
+    });
+
+    it("abandons a pending status request when the session closes and ignores its late page", async () => {
+        const status = deferred<{data: McpServerStatus[], nextCursor: string | null}>();
+        const statusParams: unknown[] = [];
+        const client = await connectSession(2, {
+            realMcpStartup: true,
+            mcpServers: [{type: "stdio", name: serverName, command: "/usr/bin/everything"} as acpV2.McpServer],
+            codexResponses: {
+                "mcpServerStatus/list": async params => {
+                    statusParams.push(params);
+                    return await status.promise;
+                },
+            },
+        });
+        clients.push(client);
+
+        emitCancelled(client);
+        await vi.waitFor(() => expect(statusParams).toHaveLength(1));
+        await expect(client.request("session/close", {sessionId})).resolves.toBeDefined();
+        expect(startupFailures(client)).toEqual([]);
+
+        status.resolve({data: [server("connected")], nextCursor: "late-page"});
+        await new Promise(resolve => setTimeout(resolve, 0));
+        expect(statusParams).toHaveLength(1);
+        expect(startupFailures(client)).toEqual([]);
+    });
+
+    it("uses only the latest complete multi-server snapshot", async () => {
+        const names = ["alpha", "beta"];
+        let calls = 0;
+        const client = await connectSession(2, {
+            realMcpStartup: true,
+            mcpServers: names.map(name => ({type: "stdio", name, command: `/usr/bin/${name}`} as acpV2.McpServer)),
+            codexResponses: {
+                "mcpServerStatus/list": async () => {
+                    calls += 1;
+                    return calls === 1
+                        ? {data: [server("connected", "alpha"), server("starting", "beta")], nextCursor: null}
+                        : {data: [server("failed", "alpha"), server("connected", "beta")], nextCursor: null};
+                },
+            },
+        });
+        clients.push(client);
+        vi.useFakeTimers();
+
+        names.forEach(name => emitCancelled(client, name));
+        await vi.advanceTimersByTimeAsync(5_000);
+
+        expect(calls).toBeGreaterThan(1);
+        const failures = startupFailures(client);
+        expect(failures).toHaveLength(1);
+        expect((failures[0] as {toolCallId?: string}).toolCallId).toContain("alpha");
+    });
+});

src/__tests__/CodexACPAgent/tool-calls-and-messages-v2.test.ts

@@ -64,6 +64,7 @@ describe('tool calls and messages over ACP v2', () => {
     let closeClient: (() => void) | null = null;
 
     afterEach(() => {
+        vi.useRealTimers();
         closeClient?.();
         closeClient = null;
         vi.clearAllMocks();
@@ -119,14 +120,22 @@ describe('tool calls and messages over ACP v2', () => {
     });
 
     it('reports MCP server startup failures as tool_call_update upserts', async () => {
+        vi.useFakeTimers();
         const client = await connectSession(2, {
             mcpServers: [
                 {type: "stdio", name: "docs", command: "/usr/bin/docs-mcp"},
                 {type: "stdio", name: "slow", command: "/usr/bin/slow-mcp"},
             ],
             mcpStartup: {ready: [], failed: [{server: "docs", error: "exit code 1"}], cancelled: ["slow"]},
+            codexResponses: {
+                "mcpServerStatus/list": async () => ({
+                    data: [{name: "slow", runtimeStatus: "cancelled"}],
+                    nextCursor: null,
+                }),
+            },
         });
         closeClient = () => client.connection.close();
+        await vi.advanceTimersByTimeAsync(5_000);
 
         const startupUpdates = () => [
             ...client.setupUpdates,

src/__tests__/CodexACPAgent/v2-prompt-harness.ts

@@ -106,6 +106,8 @@ export type PromptSession = {
     transcript: TranscriptEntry[];
     /** Updates published while the session was being set up, before `transcript` starts recording. */
     setupUpdates: acpV2.SessionUpdate[];
+    /** Every request sent to the mocked Codex app-server. */
+    codexRequests: Array<{method: string, params: unknown}>;
     turnStartParams: Array<Record<string, unknown>>;
     setTurnStart(handler: (params: Record<string, unknown>) => Promise<unknown>): void;
     /** Overrides the canned Codex response for a request method other than `turn/start`. */
@@ -135,7 +137,10 @@ export type PromptSession = {
 /** Connects a client (v2 by default) to the agent through the router, over a mocked Codex app-server, and opens a session. */
 export async function connectSession(protocolVersion: 1 | 2 = 2, options: {
     mcpServers?: acpV2.McpServer[],
+    mcpServersV1?: acp.McpServer[],
     mcpStartup?: McpStartupResult,
+    /** Use the real notification-driven startup tracker instead of a resolved test double. */
+    realMcpStartup?: boolean,
     /** Sent on `initialize` (v2 only). */
     clientCapabilities?: acpV2.ClientCapabilities,
     /** The Codex process exit code the agent sees (`null` = still running). */
@@ -153,6 +158,7 @@ export async function connectSession(protocolVersion: 1 | 2 = 2, options: {
 } = {}): Promise<PromptSession> {
     const mocks = createMockConnections();
     const transcript: TranscriptEntry[] = [];
+    const codexRequests: Array<{method: string, params: unknown}> = [];
     const turnStartParams: Array<Record<string, unknown>> = [];
     const codexResponseOverrides = new Map<string, (params: unknown) => Promise<unknown>>(
         Object.entries(options.codexResponses ?? {}),
@@ -161,6 +167,7 @@ export async function connectSession(protocolVersion: 1 | 2 = 2, options: {
         turn: createTurn("inProgress", `turn-${turnStartParams.length}`),
     });
     mocks.mockCodexConnection.sendRequest.mockImplementation(async (method: string, params?: any) => {
+        codexRequests.push({method, params});
         // The session title is generated on a separate ephemeral thread; keep it out of the way.
         if (method === "thread/start" && params?.ephemeral) {
             return {thread: {...createThread(), id: titleThreadId, ephemeral: true}};
@@ -183,7 +190,10 @@ export async function connectSession(protocolVersion: 1 | 2 = 2, options: {
     vi.spyOn(codexAcpClient, "readAuthRequirement").mockResolvedValue({required: false, account: null});
     vi.spyOn(codexAcpClient, "getAgentConfiguredModelProvider").mockResolvedValue("openai");
     vi.spyOn(codexAcpClient, "getAccount").mockResolvedValue({account: null, requiresOpenaiAuth: false});
-    vi.spyOn(codexAcpClient, "awaitMcpServerStartup").mockResolvedValue(options.mcpStartup ?? {ready: [], failed: [], cancelled: []});
+    if (!options.realMcpStartup) {
+        vi.spyOn(codexAcpClient, "awaitMcpServerStartup")
+            .mockResolvedValue(options.mcpStartup ?? {ready: [], failed: [], cancelled: []});
+    }
     let agent: CodexAcpServer | null = null;
     const router = createAcpAgentRouter((connection) => {
         agent = new CodexAcpServer(connection, codexAcpClient, undefined, options.exitCode);
@@ -248,7 +258,7 @@ export async function connectSession(protocolVersion: 1 | 2 = 2, options: {
             .onNotification(acp.methods.client.session.update, (ctx) => onUpdate(ctx.params.update))
             .connect(clientStream);
         await v1Connection.agent.request(acp.methods.agent.initialize, {protocolVersion: acp.PROTOCOL_VERSION});
-        await v1Connection.agent.request(acp.methods.agent.session.new, {cwd, mcpServers: []});
+        await v1Connection.agent.request(acp.methods.agent.session.new, {cwd, mcpServers: options.mcpServersV1 ?? []});
         connection = v1Connection;
         request = (method, params, signal) => v1Connection.agent.request(
             method as typeof acp.methods.agent.session.prompt,
@@ -288,6 +298,7 @@ export async function connectSession(protocolVersion: 1 | 2 = 2, options: {
         connection,
         transcript,
         setupUpdates,
+        codexRequests,
         turnStartParams,
         setTurnStart: (handler: typeof turnStart) => {
             turnStart = handler;

src/__tests__/scenarios/scenario-harness.ts

@@ -137,6 +137,10 @@ async function runInWorkspace(
         failed: (scenario.mcpStartup?.failed ?? []).map(failure => ({...failure, failureReason: null})),
         cancelled: scenario.mcpStartup?.cancelled ?? [],
     });
+    anyClient.listMcpServers = vi.fn().mockResolvedValue({
+        data: (scenario.mcpStartup?.cancelled ?? []).map(name => ({name, runtimeStatus: "cancelled"})),
+        nextCursor: null,
+    });
     anyAppServer.listModels = vi.fn().mockResolvedValue({data: [model], nextCursor: null});
     anyAppServer.threadBackgroundTerminalsList = vi.fn().mockResolvedValue({data: [], nextCursor: null});
     anyAppServer.threadResume = vi.fn().mockResolvedValue({
@@ -162,9 +166,16 @@ async function runInWorkspace(
         return [...recorded, ...collect(fixture.getAcpConnectionEvents([]))];
     }
 
+    if ((scenario.mcpStartup?.cancelled.length ?? 0) > 0) {
+        vi.useFakeTimers();
+    }
     const newSessionResponse = await agent.newSession({cwd: workspace, mcpServers});
     recorded.push({direction: "response", method: "session/new", params: newSessionResponse});
     if (scenario.mcpStartup !== undefined) {
+        if (scenario.mcpStartup.cancelled.length > 0) {
+            await vi.advanceTimersByTimeAsync(5_000);
+            vi.useRealTimers();
+        }
         await vi.waitFor(() => {
             const events = fixture.getAcpConnectionEvents([]);
             const count = events.filter(event => JSON.stringify(event).includes("mcp_startup")).length;

src/mcp/McpServerStatusList.ts

@@ -2,25 +2,41 @@ import type {CodexAcpClient} from "../CodexAcpClient";
 import type {McpServerStatus} from "../app-server/v2";
 import {logger} from "../Logger";
 
+interface McpServerStatusListOptions {
+    /** Stops the current read and prevents a late page from starting another request. */
+    signal?: AbortSignal;
+    /** Rejects instead of returning a partial list when pagination cannot complete. */
+    requireComplete?: boolean;
+}
+
 /**
  * Returns the status of every MCP server. With `threadId`, the status includes the live connection
  * state of that thread. The function reads all pages.
  */
-export async function listMcpServerStatus(codexAcpClient: CodexAcpClient, threadId?: string): Promise<Array<McpServerStatus>> {
+export async function listMcpServerStatus(
+    codexAcpClient: CodexAcpClient,
+    threadId?: string,
+    options: McpServerStatusListOptions = {},
+): Promise<Array<McpServerStatus>> {
     const servers: Array<McpServerStatus> = [];
     const seenCursors = new Set<string>();
     let cursor: string | null = null;
     do {
+        options.signal?.throwIfAborted();
         const page = await codexAcpClient.listMcpServers({
             cursor,
             detail: "toolsAndAuthOnly",
             ...(threadId === undefined ? {} : {threadId}),
         });
+        options.signal?.throwIfAborted();
         servers.push(...page.data);
         cursor = page.nextCursor;
         if (cursor !== null) {
             if (seenCursors.has(cursor)) {
                 logger.log(`MCP server status list repeated the cursor ${cursor}`);
+                if (options.requireComplete) {
+                    throw new Error(`MCP server status list repeated the cursor ${cursor}`);
+                }
                 break;
             }
             seenCursors.add(cursor);