feat(mcpdo): --load-connections and --load-max-* thresholds for tools/call --load

clients/mcpdo/README.md

@@ -248,11 +248,32 @@ mcpdo @prod tools/call search query:=mcp --load --load-requests 500 --load-rate
 | `--load-duration <sec>`  | Stop starting calls after this many seconds; in-flight calls finish. Default `10` without `--load-requests`. |
 | `--load-requests <n>`    | Stop after starting this many calls. With `--load-duration` too, whichever comes first ends the run.         |
 | `--load-rate <n>`        | Start at most this many calls per second, spaced evenly (default: unlimited).                                |
+| `--load-connections <n>` | Spread the calls over this many independent sessions (HTTP/SSE only; at most `--load-concurrency`; default `1`). |
+| `--load-max-error-rate <fraction>` | Exit `10` (`load_error_rate_exceeded`) when the share of failed calls is above this value (`0` to `1`). |
+| `--load-max-p99 <ms>`    | Exit `10` (`load_p99_exceeded`) when the p99 latency is above this many milliseconds, or no call completed. |
 
 Each `--load-*` flag requires `--load`, and `--load` cannot be combined with
 `--app-info`. `--format json` prints the summary object itself, with no
-`{ result }` envelope, as for every other mcpdo command. The exit code is `0`
-whenever the run completes, however many calls failed — read `errorRate`.
+`{ result }` envelope, as for every other mcpdo command. Without a threshold
+flag the exit code is `0` whenever the run completes, however many calls
+failed — read `errorRate`.
+
+**Thresholds** ([#2800](https://github.com/modelcontextprotocol/inspector/issues/2800))
+make the exit code carry the verdict, as in the one-shot CLI. They are judged
+by the `mcpdo` process after the summary has been printed, so a breached run
+still shows its numbers; the breach then follows as the usual error envelope,
+with the same `code` and exit code `10` the CLI uses.
+
+**Multiple sessions** (`--load-connections`, #2800) open the extra sessions
+inside the daemon, before the clock starts, from the connection's stored server
+config and settings. They use stored credentials only — an extra never starts a
+sign-in, so connect the connection first — and a session that fails to open is
+counted in the summary's `connections` block, not as a call error. The
+connection's own live client is always the first session. The extras belong to
+the run: they never appear in `connections/list`, they are closed when the run
+ends (or the daemon stops), and Ctrl-C cancels their in-flight calls too. Over
+stdio each session would spawn its own server process, so more than one is
+refused there.
 
 How a run behaves on a shared daemon connection:
 

clients/mcpdo/__tests__/daemon-load.test.ts

@@ -2,7 +2,13 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 import * as fs from "node:fs";
 import * as os from "node:os";
 import * as path from "node:path";
+import {
+  createEchoTool,
+  createTestServerHttp,
+  createTestServerInfo,
+} from "@modelcontextprotocol/inspector-test-server";
 import { DaemonServer } from "../src/daemon/server.js";
+import { ConnectionRegistry } from "../src/daemon/connections.js";
 import type { InspectorClient } from "@inspector/core/mcp/inspectorClient.js";
 import { InspectorClientEventTarget } from "@inspector/core/mcp/inspectorClientEventTarget.js";
 import type { ConnectionStatus } from "@inspector/core/mcp/types.js";
@@ -11,6 +17,7 @@ import type {
   LoadSummary,
 } from "@inspector/core/cli/handlers/load-test.js";
 import type { RpcParams } from "../src/daemon/protocol.js";
+import type { MCPServerConfig } from "@inspector/core/mcp/types.js";
 
 /**
  * `tools/call --load` inside the daemon (#2744): the `rpc` op hands the run to
@@ -24,7 +31,10 @@ type LoadImpl = (
   client: unknown,
   args: Record<string, unknown>,
   options: LoadOptions,
-  run: { signal?: AbortSignal },
+  run: {
+    signal?: AbortSignal;
+    openConnection?: () => Promise<InspectorClient>;
+  },
 ) => Promise<LoadSummary>;
 
 const loadMock = vi.hoisted(() => {
@@ -47,6 +57,36 @@ vi.mock(
   }),
 );
 
+/**
+ * Observes the daemon's OAuth storage handling (#2800) without changing it:
+ * the real reset runs behind a spy, and the real class is counted.
+ */
+const oauthStorageMock = vi.hoisted(() => ({
+  constructed: 0,
+  reset: vi.fn(),
+}));
+vi.mock("@inspector/core/auth/node/storage-node.js", async (importOriginal) => {
+  const actual =
+    await importOriginal<
+      typeof import("@inspector/core/auth/node/storage-node.js")
+    >();
+  oauthStorageMock.reset.mockImplementation(actual.resetNodeOAuthStorageCache);
+  return { ...actual, resetNodeOAuthStorageCache: oauthStorageMock.reset };
+});
+vi.mock("@inspector/core/auth/node/index.js", async (importOriginal) => {
+  const actual =
+    await importOriginal<typeof import("@inspector/core/auth/node/index.js")>();
+  class CountedNodeOAuthStorage extends actual.NodeOAuthStorage {
+    constructor(
+      ...args: ConstructorParameters<typeof actual.NodeOAuthStorage>
+    ) {
+      super(...args);
+      oauthStorageMock.constructed++;
+    }
+  }
+  return { ...actual, NodeOAuthStorage: CountedNodeOAuthStorage };
+});
+
 const runMethodMock = vi.hoisted(() => ({ calls: 0 }));
 vi.mock("@inspector/core/cli/handlers/run-method.js", () => ({
   runMethod: async () => {
@@ -82,7 +122,7 @@ class FakeLoadClient
   implements
     Pick<
       InspectorClient,
-      "getStatus" | "cancelToolCall" | "setAmbientRequestSignal"
+      "getStatus" | "cancelToolCall" | "setAmbientRequestSignal" | "disconnect"
     >
 {
   ambient: AbortSignal | undefined;
@@ -105,6 +145,8 @@ class FakeLoadClient
       this.ambient = undefined;
     };
   }
+
+  async disconnect(): Promise<void> {}
 }
 
 function fakeClient(status: ConnectionStatus = "connected") {
@@ -260,4 +302,224 @@ describe("daemon rpc with --load", () => {
     });
     expect(runMethodMock.calls).toBe(1);
   });
+
+  describe("--load-connections", () => {
+    const MULTI: LoadOptions = { concurrency: 2, connections: 2, requests: 4 };
+
+    /** Point the connection at a server config and stub the extra dial. */
+    function useConnection(serverConfig: MCPServerConfig) {
+      const extra = fakeClient();
+      // Only `name` and `serverConfig` are read on this path.
+      vi.spyOn(server.registry, "connectionFor").mockReturnValue({
+        name: "srv",
+        serverConfig,
+      } as ReturnType<typeof server.registry.connectionFor>);
+      const open = vi
+        .spyOn(server.registry, "openLoadClient")
+        .mockResolvedValue(extra.client);
+      const release = vi.spyOn(server.registry, "releaseLoadClient");
+      return { extra, open, release };
+    }
+
+    it("opens extras through the registry with the run's signal, and releases them", async () => {
+      const { extra, open, release } = useConnection({
+        type: "streamable-http",
+        url: "http://localhost:1/mcp",
+      });
+      let opened: InspectorClient | undefined;
+      loadMock.impl = async (_client, _args, options, run) => {
+        expect(options).toEqual(MULTI);
+        opened = await run.openConnection!();
+        // The caller hanging up aborts the extra's in-flight calls too.
+        expect(extra.ambient()).toBe(run.signal);
+        return SUMMARY;
+      };
+      const response = await rpc({ load: MULTI });
+      expect(response.ok).toBe(true);
+      expect(opened).toBe(extra.client);
+      expect(open).toHaveBeenCalledWith(
+        {
+          serverConfig: {
+            type: "streamable-http",
+            url: "http://localhost:1/mcp",
+          },
+        },
+        fake.client,
+        expect.any(AbortSignal),
+      );
+      expect(release).toHaveBeenCalledWith(extra.client);
+      expect(extra.ambient()).toBeUndefined();
+    });
+
+    it("hands the run's signal to the dial, so a stalled handshake cancels", async () => {
+      const { open } = useConnection({
+        type: "streamable-http",
+        url: "http://localhost:1/mcp",
+      });
+      let runSignal: AbortSignal | undefined;
+      loadMock.impl = async (_client, _args, _options, run) => {
+        runSignal = run.signal;
+        await run.openConnection!();
+        return SUMMARY;
+      };
+      expect((await rpc({ load: MULTI })).ok).toBe(true);
+      expect(open.mock.calls[0]![2]).toBe(runSignal);
+    });
+
+    it("dials the server the primary was resolved on, even if the default moves while queued", async () => {
+      const { open } = useConnection({
+        type: "streamable-http",
+        url: "http://a.example/mcp",
+      });
+      // A first run holds the connection's queue until its signal aborts.
+      const firstStarted = runUntilAborted();
+      const blocker = new AbortController();
+      const first = rpc({ load: LOAD }, blocker.signal);
+      await firstStarted;
+      // The second run resolves its connection now, then waits in the queue.
+      const second = rpc({ load: MULTI });
+      await vi.waitFor(() =>
+        expect(server.registry.liveClientFor).toHaveBeenCalledTimes(2),
+      );
+      // While it waits, the default connection changes to another server.
+      vi.mocked(server.registry.connectionFor).mockReturnValue({
+        name: "other",
+        serverConfig: { type: "streamable-http", url: "http://b.example/mcp" },
+      } as ReturnType<typeof server.registry.connectionFor>);
+      loadMock.impl = async (_client, _args, _options, run) => {
+        await run.openConnection!();
+        return SUMMARY;
+      };
+      blocker.abort();
+      await first;
+      expect((await second).ok).toBe(true);
+      expect(open).toHaveBeenCalledWith(
+        {
+          serverConfig: {
+            type: "streamable-http",
+            url: "http://a.example/mcp",
+          },
+        },
+        fake.client,
+        expect.any(AbortSignal),
+      );
+    });
+
+    it("gives a single-session run no way to open more", async () => {
+      const { open } = useConnection({
+        type: "streamable-http",
+        url: "http://localhost:1/mcp",
+      });
+      loadMock.impl = async (_client, _args, _options, run) => {
+        expect(run.openConnection).toBeUndefined();
+        return SUMMARY;
+      };
+      expect((await rpc({ load: { ...MULTI, connections: 1 } })).ok).toBe(true);
+      expect(open).not.toHaveBeenCalled();
+    });
+
+    it("refuses more than one session over stdio before the run starts", async () => {
+      const { open } = useConnection({ type: "stdio", command: "node" });
+      loadMock.impl = async () => SUMMARY;
+      const response = await rpc({ load: MULTI });
+      expect(response.ok).toBe(false);
+      expect(!response.ok && response.error.code).toBe("load_flag_conflict");
+      expect(open).not.toHaveBeenCalled();
+      // Nothing was set up, so nothing is left behind on the primary.
+      expect(fake.ambient()).toBeUndefined();
+    });
+  });
+});
+
+/**
+ * The registry side of `--load-connections` (#2800), against a real HTTP
+ * server: an extra session is dialed from the connection's stored config,
+ * stays out of the connection list, and is closed by a daemon shutdown that
+ * lands mid-run.
+ */
+describe("ConnectionRegistry load sessions", () => {
+  let httpServer: ReturnType<typeof createTestServerHttp>;
+  let registry: ConnectionRegistry;
+
+  beforeEach(async () => {
+    httpServer = createTestServerHttp({
+      serverInfo: createTestServerInfo("load-registry", "1.0.0"),
+      tools: [createEchoTool()],
+    });
+    await httpServer.start();
+    registry = new ConnectionRegistry(0);
+    await registry.connect({
+      name: "remote",
+      serverConfig: { type: "streamable-http", url: httpServer.url },
+      serverIdentity: httpServer.url,
+    });
+  });
+
+  afterEach(async () => {
+    await registry.disconnectAll();
+    await httpServer.stop();
+  });
+
+  const target = () => ({
+    serverConfig: { type: "streamable-http" as const, url: httpServer.url },
+  });
+  const primary = () => registry.resolve("remote", true).client;
+
+  it("dials a separate, unlisted session that shutdown closes", async () => {
+    const extra = await registry.openLoadClient(target(), primary());
+    expect(extra).not.toBe(primary());
+    expect(extra.getStatus()).toBe("connected");
+    expect(registry.list().map((c) => c.name)).toEqual(["remote"]);
+    expect(registry.connectionCount()).toBe(1);
+
+    const disconnect = vi.spyOn(extra, "disconnect");
+    await registry.disconnectAll();
+    expect(disconnect).toHaveBeenCalledOnce();
+  });
+
+  it("shares the primary's OAuth storage instead of a fresh token snapshot", async () => {
+    oauthStorageMock.reset.mockClear();
+    const constructedBefore = oauthStorageMock.constructed;
+    await registry.openLoadClient(target(), primary());
+    // A reset would orphan the primary's in-memory tokens, and a new storage
+    // would hold its own copy: either way a refresh-token rotation in one
+    // session would leave the others with an invalidated token.
+    expect(oauthStorageMock.reset).not.toHaveBeenCalled();
+    expect(oauthStorageMock.constructed).toBe(constructedBefore);
+  });
+
+  it("leaves a released session to its run", async () => {
+    const extra = await registry.openLoadClient(target(), primary());
+    await extra.disconnect();
+    registry.releaseLoadClient(extra);
+    const disconnect = vi.spyOn(extra, "disconnect");
+    await registry.disconnectAll();
+    expect(disconnect).not.toHaveBeenCalled();
+  });
+
+  it("closes a session whose dial finished after shutdown began", async () => {
+    const opening = registry.openLoadClient(target(), primary());
+    const outcome = opening.catch((err: unknown) => err);
+    await registry.disconnectAll();
+    expect(await outcome).toMatchObject({
+      envelope: { code: "daemon_stopping" },
+    });
+  });
+
+  it("refuses to open a session once the daemon is stopping", async () => {
+    const client = primary();
+    await registry.disconnectAll();
+    await expect(
+      registry.openLoadClient(target(), client),
+    ).rejects.toMatchObject({
+      envelope: { code: "daemon_stopping" },
+    });
+  });
+
+  it("cancels a dial when the run's signal aborts", async () => {
+    const stop = new AbortController();
+    const opening = registry.openLoadClient(target(), primary(), stop.signal);
+    stop.abort();
+    await expect(opening).rejects.toThrow();
+  });
 });