fix(mcp): fail in-flight calls when resume exhaustion redials

docs/MCP.md

@@ -141,10 +141,13 @@ than hanging or dispatching against a dead transport.
 
 An HTTP SSE drop does not start that second dial. The Streamable HTTP
 transport resumes the stream itself (`reconnectionOptions`: 1s start, doubling,
-30s cap). Those options have no random factor and no exhaustion callback, so
-the delay is not jittered and a stream that stops retrying stays on the
-connected row instead of flipping to failed. Tools stay live; only a real
-stdio death uses the fail-fast stubs and attempt count above.
+30s cap, 1000 attempts). Those options have no random factor, so the delay is
+not jittered; the SDK reports the end of that resume loop on the transport's
+`onerror` (`Maximum reconnection attempts (1000) exceeded.`), and Corbits
+routes exactly that terminal error to the owner's disconnect hook. The row
+flips to **reconnecting** with the standard attempt count and fail-fast stubs,
+and the owner loop redials with a fresh dial — a stream that stops retrying
+no longer sits on the connected row.
 
 Pressing Enter on a failed or reconnecting row retries now with a single dial
 instead of waiting for backoff, reusing the saved settings row — it never

src/mcp/client-resume-exhaustion.test.ts

@@ -0,0 +1,215 @@
+import { describe, expect, test } from "bun:test";
+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
+import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
+import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js";
+import {
+  armResumeExhaustionSignal,
+  disarmTransportDeathSignal,
+  isStreamableResumeExhausted,
+  type LiveHttpTransport,
+} from "./client.js";
+
+describe("isStreamableResumeExhausted", () => {
+  test.each([
+    "Maximum reconnection attempts (1000) exceeded.",
+    "Maximum reconnection attempts (1) exceeded.",
+    "Maximum reconnection attempts (999999) exceeded.",
+  ])("matches the SDK exhaustion copy (%j)", (message) => {
+    expect(isStreamableResumeExhausted(new Error(message))).toBe(true);
+  });
+
+  test.each([
+    ["a transient stream error", new Error("Failed to open SSE stream")],
+    ["a reconnect-loop error", new Error("Failed to reconnect SSE stream")],
+    [
+      "an exhaustion prefix without the tail",
+      new Error("Maximum reconnection attempts (1000)"),
+    ],
+    [
+      "an exhaustion copy without the count",
+      new Error("Maximum reconnection attempts exceeded."),
+    ],
+    ["a non-error", "Maximum reconnection attempts (1000) exceeded."],
+    ["nothing", undefined],
+  ])("rejects %s", (_label, value) => {
+    expect(isStreamableResumeExhausted(value)).toBe(false);
+  });
+});
+
+describe("armResumeExhaustionSignal", () => {
+  test("leaves the transport untouched without an owner hook", () => {
+    const transport: Pick<{ onerror?: (error: Error) => void }, "onerror"> = {};
+    armResumeExhaustionSignal(transport, () => true, undefined);
+    expect(transport.onerror).toBeUndefined();
+  });
+
+  test("routes exhaustion to the owner hook and keeps the prior handler", () => {
+    const seen: string[] = [];
+    let disconnects = 0;
+    const transport: { onerror?: (error: Error) => void } = {
+      onerror: (error) => {
+        seen.push(error.message);
+      },
+    };
+    armResumeExhaustionSignal(
+      transport,
+      () => true,
+      () => {
+        disconnects += 1;
+      },
+    );
+    transport.onerror?.(
+      new Error("Maximum reconnection attempts (1000) exceeded."),
+    );
+    expect(seen).toEqual(["Maximum reconnection attempts (1000) exceeded."]);
+    expect(disconnects).toBe(1);
+  });
+
+  test("ignores non-exhaustion transport errors", () => {
+    let disconnects = 0;
+    const transport: { onerror?: (error: Error) => void } = {};
+    armResumeExhaustionSignal(
+      transport,
+      () => true,
+      () => {
+        disconnects += 1;
+      },
+    );
+    transport.onerror?.(new Error("Failed to open SSE stream"));
+    expect(disconnects).toBe(0);
+  });
+
+  test("stays silent for a transport superseded by a recovery redial", () => {
+    let disconnects = 0;
+    const seen: string[] = [];
+    const transport: { onerror?: (error: Error) => void } = {
+      onerror: (error) => {
+        seen.push(error.message);
+      },
+    };
+    armResumeExhaustionSignal(
+      transport,
+      () => false,
+      () => {
+        disconnects += 1;
+      },
+    );
+    transport.onerror?.(
+      new Error("Maximum reconnection attempts (1000) exceeded."),
+    );
+    expect(seen).toEqual(["Maximum reconnection attempts (1000) exceeded."]);
+    expect(disconnects).toBe(0);
+  });
+});
+
+describe("disarmTransportDeathSignal", () => {
+  test("keeps protocol onclose and stays silent for the owner", () => {
+    let disconnects = 0;
+    let protocolCloses = 0;
+    const live: LiveHttpTransport = {};
+    const transport: {
+      onclose?: () => void;
+      onerror?: (error: Error) => void;
+    } = {
+      onclose: () => {
+        protocolCloses += 1;
+      },
+    };
+    live.current = transport;
+    armResumeExhaustionSignal(
+      transport,
+      () => live.current === transport && live.disarmed !== true,
+      () => {
+        disconnects += 1;
+      },
+    );
+
+    disarmTransportDeathSignal(live);
+
+    live.current?.onclose?.();
+    live.current?.onerror?.(
+      new Error("Maximum reconnection attempts (1000) exceeded."),
+    );
+    expect(disconnects).toBe(0);
+    expect(protocolCloses).toBe(1);
+    expect(live.disarmed).toBe(true);
+  });
+
+  test("tolerates an empty handle", () => {
+    expect(() => disarmTransportDeathSignal({})).not.toThrow();
+  });
+});
+
+const RESUME_EXHAUSTED = "Maximum reconnection attempts (1000) exceeded.";
+
+/** Transport that answers initialize and then never completes a request. */
+class HungToolsTransport implements Transport {
+  onclose?: () => void;
+  onerror?: (error: Error) => void;
+  onmessage?: (message: JSONRPCMessage) => void;
+
+  async start(): Promise<void> {
+    return undefined;
+  }
+
+  async close(): Promise<void> {
+    this.onclose?.();
+  }
+
+  async send(message: JSONRPCMessage): Promise<void> {
+    if (!("method" in message) || !("id" in message)) return;
+    if (message.method !== "initialize") return;
+    this.onmessage?.({
+      jsonrpc: "2.0",
+      id: message.id,
+      result: {
+        protocolVersion: "2025-11-25",
+        capabilities: { tools: {} },
+        serverInfo: { name: "hung", version: "0" },
+      },
+    });
+  }
+}
+
+describe("resume exhaustion closes in-flight SDK requests", () => {
+  test("rejects listTools when the SDK reports resume exhaustion", async () => {
+    const client = new Client({ name: "resume-exhaustion", version: "0" });
+    const transport = new HungToolsTransport();
+    const live: LiveHttpTransport = {};
+    let disconnects = 0;
+    armResumeExhaustionSignal(
+      transport,
+      () => live.current === transport && live.disarmed !== true,
+      () => {
+        disconnects += 1;
+        // Owner redial: MCPClient.close disarms, then Protocol.close.
+        disarmTransportDeathSignal(live);
+        void client.close();
+      },
+    );
+    live.current = transport;
+    await client.connect(transport);
+
+    const pending = client.listTools();
+    transport.onerror?.(new Error(RESUME_EXHAUSTED));
+
+    const outcome = await Promise.race([
+      pending.then(
+        () => "resolved" as const,
+        (err: unknown) => err,
+      ),
+      new Promise<"pending">((resolve) => {
+        setTimeout(() => resolve("pending"), 500);
+      }),
+    ]);
+
+    expect(disconnects).toBe(1);
+    expect(outcome).not.toBe("pending");
+    expect(outcome).not.toBe("resolved");
+    expect(outcome).toBeInstanceOf(Error);
+    if (!(outcome instanceof Error)) return;
+    expect(outcome.message).toBe("MCP error -32000: Connection closed");
+    expect("code" in outcome && outcome.code).toBe(-32000);
+    expect(client.transport).toBeUndefined();
+  });
+});

src/mcp/client.ts

@@ -87,9 +87,11 @@ export interface MCPConnectOptions {
    */
   onAuthorized?: (serverName: string) => void;
   signal?: AbortSignal;
-  // Wired to transport.onclose after connect so the owner learns the
-  // transport died under a live client; close() disarms it before
-  // intentional teardown.
+  // Wired to transport.onclose after connect, plus HTTP SSE-resume
+  // exhaustion via transport.onerror, so the owner learns the transport
+  // died under a live client. close() marks the live handle disarmed
+  // before intentional teardown; it must not replace onclose, or
+  // Protocol._onclose never rejects in-flight calls.
   onDisconnect?: () => void;
   // Session OS-sandbox handle: a model-reachable stdio server spawn runs
   // wrapped; an unwrappable command fails the connect (fail closed). Omitted
@@ -406,6 +408,62 @@ export const MCP_HTTP_RECONNECTION_OPTIONS = {
   maxRetries: 1_000,
 } as const;
 
+/** Live HTTP transport handle shared with finishClient so intentional
+ * teardown can suppress the owner death signal without replacing
+ * `onclose` (Protocol._onclose rejects in-flight calls from that hook). */
+export interface LiveHttpTransport {
+  current?: {
+    onclose?: () => void;
+    onerror?: (error: Error) => void;
+  };
+  /** Intentional teardown. Owner hooks must consult this and stay quiet. */
+  disarmed?: boolean;
+}
+
+/**
+ * SDK SSE resume reports a dead stream on `transport.onerror`, never
+ * `onclose`: once the reconnection cap trips it emits
+ * `Maximum reconnection attempts (N) exceeded.` and stops. Without a hook
+ * the row would stay on the connected row forever, so exactly that terminal
+ * error routes to the owner's disconnect hook. The copy is pinned to the
+ * ^1.32.1 SDK range in package.json.
+ */
+const RESUME_EXHAUSTED_PATTERN =
+  /^Maximum reconnection attempts \(\d+\) exceeded\.$/;
+
+export function isStreamableResumeExhausted(error: unknown): boolean {
+  return error instanceof Error && RESUME_EXHAUSTED_PATTERN.test(error.message);
+}
+
+/**
+ * Chain SSE-resume exhaustion onto the owner's disconnect hook so the row
+ * flips to reconnecting and the owner loop redials with a fresh dial.
+ * Stale transports (superseded by a recovery redial) stay silent; every
+ * other transport error passes through to the prior handler untouched.
+ */
+export function armResumeExhaustionSignal(
+  transport: Pick<StreamableHTTPClientTransport, "onerror">,
+  isLive: () => boolean,
+  onDisconnect?: () => void,
+): void {
+  if (onDisconnect === undefined) return;
+  const prior = transport.onerror;
+  transport.onerror = (error: Error) => {
+    prior?.(error);
+    if (isLive() && isStreamableResumeExhausted(error)) onDisconnect();
+  };
+}
+
+/** Suppress the owner death signal for intentional teardown.
+ * Do not replace `onclose`: Protocol.close only calls transport.close(),
+ * and the SDK rejects in-flight requests from the onclose wrapper
+ * installed at connect (`_onclose`, MCP -32000 Connection closed). A noop
+ * there leaves the HTTP RPC pending until the request timeout.
+ */
+export function disarmTransportDeathSignal(live: LiveHttpTransport): void {
+  live.disarmed = true;
+}
+
 export function streamableHTTPTransportOptions(
   authProvider: CorbitsOAuthProvider | undefined,
   signal: AbortSignal | undefined,
@@ -644,7 +702,7 @@ async function finishClient(
   authContext?: HTTPAuthContext,
   signal?: AbortSignal,
   closeLifecycle?: () => void,
-  live?: { current?: { onclose?: () => void } },
+  live?: LiveHttpTransport,
 ): Promise<MCPClient> {
   const result = await withHTTPAuthorizationRecovery(authContext, () =>
     signal === undefined
@@ -693,7 +751,7 @@ async function finishClient(
     },
     async close() {
       // Disarm first: the owner's hook reports unintentional death only.
-      if (live?.current !== undefined) live.current.onclose = () => undefined;
+      if (live !== undefined) disarmTransportDeathSignal(live);
       closeLifecycle?.();
       authContext?.callback.close();
       await closeQuietly(() => client.close());
@@ -746,21 +804,23 @@ async function connectStdio(
   const client = new Client({ name: MCP_CLIENT_NAME, version: "1.0.0" });
   try {
     const transport = new StdioClientTransport(transportOptions);
+    const live: LiveHttpTransport = { current: transport };
     await client.connect(
       transport,
       options.signal === undefined ? undefined : { signal: options.signal },
     );
     // Child exit is the only death the owner redials. Chain the protocol
     // handler so in-flight requests still reject; HTTP SSE resume stays
-    // inside the SDK and must not start a second dial.
+    // inside the SDK and must not start a second dial. Disarm skips the
+    // owner hook only — prior() is Protocol._onclose and must still run.
     if (options.onDisconnect !== undefined) {
       const prior = transport.onclose;
       transport.onclose = () => {
         prior?.();
+        if (live.disarmed === true) return;
         options.onDisconnect?.();
       };
     }
-    const live = { current: transport };
     return {
       ok: true,
       client: await finishClient(
@@ -807,7 +867,7 @@ async function connectHttp(
     const normalizedURL = normalizeMCPServerURL(config.url);
     const url = new URL(normalizedURL);
     let authContext: HTTPAuthContext | undefined;
-    const live: { current?: { onclose?: () => void } } = {};
+    const live: LiveHttpTransport = {};
     const connectedClient = new Client({
       name: MCP_CLIENT_NAME,
       version: "1.0.0",
@@ -820,6 +880,11 @@ async function connectHttp(
           url,
           streamableHTTPTransportOptions(undefined, options.signal),
         );
+        armResumeExhaustionSignal(
+          transport,
+          () => live.current === transport && live.disarmed !== true,
+          options.onDisconnect,
+        );
         live.current = transport;
         return transport as unknown as Transport;
       };
@@ -841,6 +906,11 @@ async function connectHttp(
           url,
           streamableHTTPTransportOptions(authProvider, lifecycle.signal),
         );
+        armResumeExhaustionSignal(
+          transport,
+          () => live.current === transport && live.disarmed !== true,
+          options.onDisconnect,
+        );
         coordinator.transport = transport;
         live.current = transport;
         return transport as unknown as Transport;