Retry failed MCP connects and abort OAuth HTTP

Failed connects used to stay reserved so same-session retry looked like an active duplicate. hasMCPServer now means connected or in-flight. Streamable HTTP fetch always composes the connect AbortSignal so SDK 403 auth() aborts.

Author: TheGreatAxios

src/agent/exa-web-fetch-alias.test.ts

@@ -5,7 +5,7 @@ import { join } from "node:path";
 import type { ToolResult } from "@intx/types/runtime";
 import { stringTool, type AgentTool } from "@intx/agent";
 import { withMockedModule } from "../../tests/helpers/mock-module.js";
-import type { ResolvedMCPServerConfig } from "../mcp/exa.js";
+import { createExaMCPServerConfig, type ResolvedMCPServerConfig } from "../mcp/exa.js";
 import type { MCPConnectOptions } from "../mcp/client.js";
 import { createGlobalSettingsWriter, persistGlobalHTTPMCPServer } from "../mcp/add-server.js";
 import { createPermissionGate } from "../permission/gate.js";
@@ -395,13 +395,13 @@ describe("built-in Exa web_fetch alias", () => {
     }
   });
 
-  test("failed implicit Exa remains reserved and cannot be persisted explicitly", async () => {
+  test("failed implicit Exa is not active and retries without a second persist", async () => {
     connectMode = "failed";
     const toolset = await makeToolset();
     const path = join(mkdtempSync(join(tmpdir(), "corbits-mcp-failed-exa-")), "settings.json");
     try {
       await connect(toolset);
-      expect(toolset.hasMCPServer("exa")).toBe(true);
+      expect(toolset.hasMCPServer("exa")).toBe(false);
       expect(
         await persistGlobalHTTPMCPServer(
           createGlobalSettingsWriter(path),
@@ -410,8 +410,15 @@ describe("built-in Exa web_fetch alias", () => {
           "none",
           toolset.hasMCPServer,
         ),
-      ).toEqual({ ok: false, reason: "active" });
-      expect(await Bun.file(path).exists()).toBe(false);
+      ).toMatchObject({ ok: true, server: { name: "exa" } });
+
+      connectMode = "success";
+      await toolset.connectMCPServer(createExaMCPServerConfig(), {
+        interactiveAuth: false,
+        onStatus: () => undefined,
+        onToolsChanged: () => undefined,
+      });
+      expect(toolset.hasMCPServer("exa")).toBe(true);
     } finally {
       await toolset.dispose();
     }
@@ -477,7 +484,7 @@ describe("built-in Exa web_fetch alias", () => {
       expect(
         toolset.dynamicRunner.currentDefinitions().some((tool) => tool.name.includes("linear")),
       ).toBe(false);
-      expect(toolset.hasMCPServer("linear")).toBe(true);
+      expect(toolset.hasMCPServer("linear")).toBe(false);
 
       connectMode = "failed";
       const retryStates: string[] = [];
@@ -518,6 +525,29 @@ describe("built-in Exa web_fetch alias", () => {
 
       expect(states.map((status) => status.state)).toEqual(["connecting", "failed"]);
       expect(states[1]?.error).toContain("connection exploded");
+      expect(toolset.hasMCPServer("linear")).toBe(false);
+      expect(await Bun.file(path).json()).toMatchObject({
+        mcpServers: [{ name: "linear", type: "http", url: "https://mcp.linear.app/mcp" }],
+      });
+      expect(
+        await persistGlobalHTTPMCPServer(
+          createGlobalSettingsWriter(path),
+          "linear",
+          "https://mcp.linear.app/mcp",
+          "none",
+          toolset.hasMCPServer,
+        ),
+      ).toEqual({ ok: false, reason: "duplicate" });
+
+      connectMode = "success";
+      const retryStates: string[] = [];
+      await toolset.connectMCPServer(persisted.server, {
+        interactiveAuth: true,
+        onStatus: (status) => retryStates.push(status.state),
+        onToolsChanged: () => undefined,
+      });
+      expect(retryStates).toEqual(["connecting", "connected"]);
+      expect(toolset.hasMCPServer("linear")).toBe(true);
       expect(await Bun.file(path).json()).toMatchObject({
         mcpServers: [{ name: "linear", type: "http", url: "https://mcp.linear.app/mcp" }],
       });

src/agent/tools.ts

@@ -209,6 +209,9 @@ export interface AgentToolset {
     callbacks: MCPConnectCallbacks,
     signal?: AbortSignal,
   ) => Promise<void>;
+  // True while this name is connected or a connection is in flight — not after
+  // a failed connect. Persist uses this to block a second add of an active name;
+  // failed rows retry through connectMCPServer without a second persist.
   hasMCPServer: (name: string) => boolean;
   // Wire the callback the `tool_search` tool invokes to make matched tools
   // advertised. Set by the runner once the director + reload loop exist.
@@ -527,7 +530,6 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
   const dynamicRunner = createDynamicToolRunner(primaryTools, toolWatchdog);
   runnerHolder.current = dynamicRunner;
 
-  const reservedMCPServerNames = new Set(mcpServers.map((server) => server.name));
   const connectedClients = new Map<string, MCPClient>();
   const inFlightConnections = new Map<string, Promise<void>>();
   const mcpAbortController = new AbortController();
@@ -540,7 +542,6 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
     signal?: AbortSignal,
   ): Promise<void> => {
     if (disposed) return Promise.resolve();
-    reservedMCPServerNames.add(config.name);
     if (connectedClients.has(config.name)) return Promise.resolve();
     const existing = inFlightConnections.get(config.name);
     if (existing !== undefined) return existing;
@@ -691,7 +692,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
     dynamicRunner,
     connectMCP,
     connectMCPServer: connectOneMCPServer,
-    hasMCPServer: (name) => reservedMCPServerNames.has(name),
+    hasMCPServer: (name) => connectedClients.has(name) || inFlightConnections.has(name),
     setToolPromoter: (promote) => {
       promoter.promote = promote;
     },

src/mcp/client-auth-policy.test.ts

@@ -82,7 +82,10 @@ await withMockedModule(
     StreamableHTTPClientTransport: class {
       constructor(
         _url: URL,
-        private readonly options?: { requestInit?: RequestInit },
+        private readonly options?: {
+          requestInit?: RequestInit;
+          fetch?: (url: string | URL, init?: RequestInit) => Promise<Response>;
+        },
       ) {
         transportOptions.push(options);
         lastTransportAuth = () => this.auth();
@@ -101,6 +104,9 @@ await withMockedModule(
         }
       }
       async auth(): Promise<void> {
+        // SDK 403 upscoping uses raw `_fetch` with no init.signal. Hang on the
+        // connect signal the product also installs as `fetch`, so abort still
+        // settles this path.
         const signal = this.options?.requestInit?.signal;
         tokenRefreshSignals.push(signal);
         await hangUntilAbort(
@@ -149,7 +155,7 @@ await withMockedModule(
   }),
 );
 
-const { connectMCPServer } = await import("./client.js");
+const { connectMCPServer, fetchWithConnectAbort } = await import("./client.js");
 
 describe("HTTP MCP auth policy", () => {
   beforeEach(() => {
@@ -257,14 +263,24 @@ describe("HTTP MCP auth policy", () => {
     );
     expect(result.ok).toBe(true);
     if (!result.ok) return;
-    expect(transportOptions).toEqual([{ authProvider, requestInit: { signal: abort.signal } }]);
+    expect(transportOptions).toEqual([
+      { authProvider, requestInit: { signal: abort.signal }, fetch: expect.any(Function) },
+    ]);
 
     const call = result.client.call("ping", {}, abort.signal);
     while (tokenRefreshSignals.length === 0) await Promise.resolve();
     expect(tokenRefreshSignals[0]).toBe(abort.signal);
     abort.abort(new Error("toolset disposed"));
     await expect(call).rejects.toThrow("toolset disposed");
     expect(tokenRefreshAborts).toBe(1);
+
+    const fetchFn = (
+      transportOptions[0] as {
+        fetch?: (url: string | URL, init?: RequestInit) => Promise<Response>;
+      }
+    ).fetch;
+    expect(fetchFn).toBeTypeOf("function");
+    await expect(fetchFn!("https://auth.test/token")).rejects.toThrow();
   });
 
   test("ordinary HTTP creates endpoint-scoped OAuth and passes it to transport", async () => {
@@ -281,3 +297,40 @@ describe("HTTP MCP auth policy", () => {
     expect(transportOptions).toEqual([{ authProvider }]);
   });
 });
+
+describe("fetchWithConnectAbort", () => {
+  test("rejects when the connect signal aborts OAuth HTTP with no init.signal", async () => {
+    const abort = new AbortController();
+    let seen: AbortSignal | undefined;
+    const fetchFn = fetchWithConnectAbort(abort.signal, (_url, init) => {
+      seen = init?.signal ?? undefined;
+      return hangUntilAbort(init?.signal, () => undefined, "aborted").then(
+        () => new Response(null, { status: 200 }),
+      );
+    });
+
+    const pending = fetchFn("https://auth.test/token");
+    expect(seen).toBe(abort.signal);
+    abort.abort(new Error("toolset disposed"));
+    await expect(pending).rejects.toThrow("toolset disposed");
+  });
+
+  test("composes connect abort with a caller request signal", async () => {
+    const connect = new AbortController();
+    const request = new AbortController();
+    let seen: AbortSignal | undefined;
+    const fetchFn = fetchWithConnectAbort(connect.signal, (_url, init) => {
+      seen = init?.signal ?? undefined;
+      return hangUntilAbort(init?.signal, () => undefined, "aborted").then(
+        () => new Response(null, { status: 200 }),
+      );
+    });
+
+    const pending = fetchFn("https://auth.test/token", { signal: request.signal });
+    expect(seen).toBeDefined();
+    expect(seen).not.toBe(connect.signal);
+    expect(seen).not.toBe(request.signal);
+    connect.abort(new Error("toolset disposed"));
+    await expect(pending).rejects.toThrow("toolset disposed");
+  });
+});

src/mcp/client.ts

@@ -72,14 +72,35 @@ function isRecoverableAuthError(err: unknown): boolean {
   return err instanceof UnauthorizedError || err instanceof OAuthError;
 }
 
+/**
+ * Fetch that always attaches the connect AbortSignal. SDK 403 upscoping calls
+ * `auth()` with raw `_fetch` (no `requestInit.signal`); `_fetchWithInit` still
+ * uses this same function, so both paths abort when connect is cancelled.
+ */
+export function fetchWithConnectAbort(
+  connectSignal: AbortSignal,
+  baseFetch: (url: string | URL, init?: RequestInit) => Promise<Response> = fetch,
+): (url: string | URL, init?: RequestInit) => Promise<Response> {
+  return (url, init) => {
+    const requestSignal = init?.signal ?? undefined;
+    const signal =
+      requestSignal === undefined || requestSignal === connectSignal
+        ? connectSignal
+        : AbortSignal.any([connectSignal, requestSignal]);
+    return baseFetch(url, { ...init, signal });
+  };
+}
+
 function streamableHTTPTransportOptions(
   authProvider: CorbitsOAuthProvider | undefined,
   signal: AbortSignal | undefined,
 ) {
   if (authProvider === undefined && signal === undefined) return undefined;
   return {
     ...(authProvider === undefined ? {} : { authProvider }),
-    ...(signal === undefined ? {} : { requestInit: { signal } }),
+    ...(signal === undefined
+      ? {}
+      : { requestInit: { signal }, fetch: fetchWithConnectAbort(signal) }),
   };
 }