diff --git a/clients/cli/README.md b/clients/cli/README.md index 8db062f49..da58f8a08 100644 --- a/clients/cli/README.md +++ b/clients/cli/README.md @@ -301,6 +301,8 @@ Exit codes are unchanged: a `tools/call` that returns `isError:true` is still wr all — at one tool and keeps calling it, so seeing how a server behaves under concurrent or sustained traffic does not mean rewriting that setup in a separate script ([#2439](https://github.com/modelcontextprotocol/inspector/issues/2439)). +`mcpdo` offers the same mode against an open connection — see +[`tools/call --load`](../mcpdo/README.md#load-testing-a-tool-toolscall---load). ```bash # 20 calls in flight for 30 seconds diff --git a/clients/cli/__tests__/load-test.test.ts b/clients/cli/__tests__/load-test.test.ts index a7ffb6684..c2e84df77 100644 --- a/clients/cli/__tests__/load-test.test.ts +++ b/clients/cli/__tests__/load-test.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect } from "vitest"; +import { describe, it, expect, vi } from "vitest"; import { runCli } from "./helpers/cli-runner.js"; import { expectCliSuccess, @@ -394,6 +394,106 @@ describe("load-test engine", () => { expect(summary.elapsedMs).toBe(1500); }); + it("starts no further call once its signal aborts", async () => { + const abort = new AbortController(); + let calls = 0; + const call = async (): Promise => { + if (++calls === 3) abort.abort(); + return "ok"; + }; + // A rate, so the waits between those calls end on their timers. + const summary = await runLoad( + call, + { concurrency: 1, requests: 100, rate: 1000 }, + fakeClock(), + abort.signal, + ); + expect(summary.requests).toBe(3); + }); + + it("cuts a rate-limit wait short when its signal aborts", async () => { + const abort = new AbortController(); + // A timer that never fires: only the abort can end the wait. + const clock: LoadClock = { + now: () => 1000, + sleep: () => new Promise(() => {}), + }; + const run = runLoad( + async () => "ok", + { concurrency: 1, requests: 5, rate: 1 }, + clock, + abort.signal, + ); + await new Promise((r) => setImmediate(r)); + abort.abort(); + expect((await run).requests).toBe(1); + }); + + it("does not run a claimed slot once another worker aborted", async () => { + const abort = new AbortController(); + let calls = 0; + const call = async (): Promise => { + calls++; + abort.abort(); + return "ok"; + }; + await runLoad( + call, + { concurrency: 2, requests: 10 }, + fakeClock(), + abort.signal, + ); + expect(calls).toBe(1); + }); + + it("drops the real clock's rate timer when the run aborts", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + try { + const abort = new AbortController(); + // Default (real) clock: the second slot is 1000s away. + const run = runLoad( + async () => "ok", + { concurrency: 1, requests: 2, rate: 0.001 }, + undefined, + abort.signal, + ); + await vi.waitFor(() => expect(vi.getTimerCount()).toBe(1)); + abort.abort(); + expect((await run).requests).toBe(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("lets the event loop run while calls fail instantly", async () => { + const abort = new AbortController(); + // Only a macrotask can abort: if the run never yields, this never fires + // before the run has spent its whole request count. + setImmediate(() => abort.abort()); + const limit = 200_000; + const summary = await runLoad( + async () => { + throw new Error("rejected up front"); + }, + { concurrency: 4, requests: limit }, + undefined, + abort.signal, + ); + expect(summary.requests).toBeGreaterThan(0); + expect(summary.requests).toBeLessThan(limit); + }); + + it("starts nothing when its signal has already aborted", async () => { + const summary = await runLoad( + async () => "ok", + { concurrency: 2, requests: 5 }, + fakeClock(), + AbortSignal.abort(), + ); + expect(summary.requests).toBe(0); + }); + it("reports an empty run without dividing by zero", async () => { const summary = await runLoad( async () => "ok", diff --git a/clients/mcpdo/README.md b/clients/mcpdo/README.md index 3b9747927..8e27ce311 100644 --- a/clients/mcpdo/README.md +++ b/clients/mcpdo/README.md @@ -74,6 +74,7 @@ mcpdo auth/clear --all --yes mcpdo tools/list mcpdo tools/call echo message:=hi mcpdo tools/call echo '{"message":"hi"}' +mcpdo tools/call echo message:=hi --load --load-requests 500 # load test; see below mcpdo @test-stdio resources/list mcpdo logging/tail # long-lived; Ctrl-C to stop mcpdo connections/list @@ -221,6 +222,62 @@ mcpdo connect my-server --config path/to/mcp.json --elicit off mcpdo connect https://example.com/mcp --elicit url ``` +## Load testing a tool (`tools/call --load`) + +`tools/call --load` is the one-shot CLI's +[load mode](../cli/README.md#load-testing-a-tool---load), run against an open +connection ([#2744](https://github.com/modelcontextprotocol/inspector/issues/2744)). +It calls one tool repeatedly and concurrently and prints a single summary in +place of a result: `ok`, `toolErrors`, `errors`, `errorRate`, `throughput`, +latency `min`/`mean`/`p50`/`p90`/`p99`/`max`, and the most frequent error +messages. Both CLIs run the same engine, so the flags, their defaults and the +summary fields mean the same thing in each. + +```bash +# 20 calls in flight for 30 seconds +mcpdo tools/call search query:=mcp --load --load-concurrency 20 --load-duration 30 + +# exactly 500 calls, started at no more than 50 per second, as JSON +mcpdo @prod tools/call search query:=mcp --load --load-requests 500 --load-rate 50 --format json +``` + +| Flag | Meaning | +| ------------------------ | ------------------------------------------------------------------------------------------------------------ | +| `--load` | Load-test the tool instead of calling it once. | +| `--load-concurrency ` | Most calls in flight at once (positive integer; default `10`). | +| `--load-duration ` | Stop starting calls after this many seconds; in-flight calls finish. Default `10` without `--load-requests`. | +| `--load-requests ` | Stop after starting this many calls. With `--load-duration` too, whichever comes first ends the run. | +| `--load-rate ` | Start at most this many calls per second, spaced evenly (default: unlimited). | + +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`. + +How a run behaves on a shared daemon connection: + +- **The run happens inside the daemon**, on the connection's live client — its + transport, headers and OAuth tokens — and the `mcpdo` process makes one round + trip for the whole run. The IPC call has no client-side deadline, so a long + `--load-duration` is not cut off; the connection's own request timeout still + applies to each call. +- **Later commands on the same connection wait for it.** The daemon runs one + `rpc` command per connection at a time (so that an elicitation can be routed + to the call that raised it), and the load run is one such command. A + `tools/list` issued mid-run starts when the run ends. Other connections, and + an already-running `logging/tail` or `resources/subscribe` stream, are + unaffected. +- **Ctrl-C ends the run**: no further calls start and those in flight are + cancelled, so the connection is free for the next command at once. No summary + is printed, since the caller has gone. +- **A dropped connection ends the run early** with the summary of what + completed, rather than spending the rest of the window counting instant + failures. The next command revives the connection as usual. +- **Elicitations are cancelled.** Many calls are in flight at once and none of + them can be answered mid-run, so a server that elicits during a load run sees + each request cancelled. Use `connect --elicit off` to keep the server from + asking at all. + ## Relation to one-shot CLI | | One-shot | Connection (`mcpdo`) | diff --git a/clients/mcpdo/__tests__/daemon-load.test.ts b/clients/mcpdo/__tests__/daemon-load.test.ts new file mode 100644 index 000000000..ade7f3534 --- /dev/null +++ b/clients/mcpdo/__tests__/daemon-load.test.ts @@ -0,0 +1,263 @@ +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 { DaemonServer } from "../src/daemon/server.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"; +import type { + LoadOptions, + LoadSummary, +} from "@inspector/core/cli/handlers/load-test.js"; +import type { RpcParams } from "../src/daemon/protocol.js"; + +/** + * `tools/call --load` inside the daemon (#2744): the `rpc` op hands the run to + * the shared engine against the connection's live client and returns its + * summary, and the run ends early when its caller hangs up or its connection + * drops. The engine itself is covered by the CLI's load-test suite; here it + * is stubbed so each test controls when the run settles. + */ + +type LoadImpl = ( + client: unknown, + args: Record, + options: LoadOptions, + run: { signal?: AbortSignal }, +) => Promise; + +const loadMock = vi.hoisted(() => { + const state: { impl: LoadImpl } = { + // A test that forgets to set `impl` fails here, by name. + impl: async () => { + throw new Error("runToolLoad mock not configured"); + }, + }; + return state; +}); +vi.mock( + "@inspector/core/cli/handlers/load-test.js", + async (importOriginal) => ({ + ...(await importOriginal< + typeof import("@inspector/core/cli/handlers/load-test.js") + >()), + runToolLoad: (...args: Parameters) => + loadMock.impl(...args), + }), +); + +const runMethodMock = vi.hoisted(() => ({ calls: 0 })); +vi.mock("@inspector/core/cli/handlers/run-method.js", () => ({ + runMethod: async () => { + runMethodMock.calls++; + return { kind: "result", result: {} }; + }, +})); + +const SUMMARY: LoadSummary = { + tool: "echo", + concurrency: 2, + requestLimit: 4, + requests: 4, + ok: 4, + toolErrors: 0, + errors: 0, + errorRate: 0, + elapsedMs: 10, + throughput: 400, + latencyMs: { min: 1, mean: 2, p50: 2, p90: 3, p99: 3, max: 3 }, + errorMessages: [], +}; + +const LOAD: LoadOptions = { concurrency: 2, requests: 4 }; + +/** + * The slice of `InspectorClient` the load path touches, on the real event + * target so `statusChange` is dispatched exactly as the client does it. + * `implements` checks each method against the client's own signature. + */ +class FakeLoadClient + extends InspectorClientEventTarget + implements + Pick< + InspectorClient, + "getStatus" | "cancelToolCall" | "setAmbientRequestSignal" + > +{ + ambient: AbortSignal | undefined; + + constructor(private readonly status: ConnectionStatus) { + super(); + } + + getStatus(): ConnectionStatus { + return this.status; + } + + cancelToolCall(): boolean { + return false; + } + + setAmbientRequestSignal(signal: AbortSignal | undefined): () => void { + this.ambient = signal; + return () => { + this.ambient = undefined; + }; + } +} + +function fakeClient(status: ConnectionStatus = "connected") { + const fake = new FakeLoadClient(status); + return { + // The daemon is typed against the concrete `InspectorClient` class, whose + // private members no test double can supply, so no single cast relates + // the two. The double is sufficient because the load path calls only the + // methods `FakeLoadClient` implements (checked above) plus the event + // target it inherits; `runToolLoad`, which needs the rest, is mocked. + client: fake as unknown as InspectorClient, + ambient: () => fake.ambient, + emitStatus: (detail: ConnectionStatus) => + fake.dispatchTypedEvent("statusChange", detail), + }; +} + +describe("daemon rpc with --load", () => { + let dir: string; + let server: DaemonServer; + let fake: ReturnType; + + // Only `liveClientFor` needs replacing: a load run never parks, and the + // ordinary call below sends no `interactive`, so neither path resolves a + // connection name through `connectionFor`. + function useClient(status?: ConnectionStatus) { + fake = fakeClient(status); + vi.spyOn(server.registry, "liveClientFor").mockResolvedValue(fake.client); + } + + beforeEach(() => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), "mcp-load-")); + server = new DaemonServer({ dir, idleMs: 0 }); + runMethodMock.calls = 0; + useClient(); + }); + + afterEach(() => { + fs.rmSync(dir, { recursive: true, force: true }); + vi.restoreAllMocks(); + }); + + function rpc(params: Partial, signal?: AbortSignal) { + return server.handle( + { + id: "r1", + op: "rpc", + params: { + method: "tools/call", + name: "srv", + toolName: "echo", + load: LOAD, + ...params, + }, + }, + undefined, + signal, + ); + } + + /** Start a run that settles only when its signal aborts. */ + function runUntilAborted(): Promise { + return new Promise((started) => { + loadMock.impl = (_client, _args, _options, run) => { + started(run.signal!); + return new Promise((resolve) => { + run.signal!.addEventListener("abort", () => + resolve({ ...SUMMARY, requests: 1 }), + ); + }); + }; + }); + } + + it("runs the engine on the live client and returns its summary", async () => { + let seen: { args: Record; options: LoadOptions } | null = + null; + loadMock.impl = async (client, args, options, run) => { + expect(client).toBe(fake.client); + // The run's own signal is the ambient one, so in-flight calls abort too. + expect(fake.ambient()).toBe(run.signal); + seen = { args, options }; + return SUMMARY; + }; + const response = await rpc({ interactive: true }); + expect(response).toEqual({ + id: "r1", + ok: true, + result: { kind: "load", summary: SUMMARY }, + }); + expect(seen!.options).toEqual(LOAD); + // Routing fields stay daemon-side; only method arguments reach the engine. + expect(seen!.args).toEqual({ method: "tools/call", toolName: "echo" }); + expect(fake.ambient()).toBeUndefined(); + expect(runMethodMock.calls).toBe(0); + }); + + it("runs rather than parks for a non-interactive caller", async () => { + loadMock.impl = async () => SUMMARY; + const response = await rpc({ interactive: false }); + expect(response.ok && response.result).toEqual({ + kind: "load", + summary: SUMMARY, + }); + }); + + it("ends the run when the caller hangs up", async () => { + const abort = new AbortController(); + const started = runUntilAborted(); + const call = rpc({}, abort.signal); + const runSignal = await started; + expect(runSignal.aborted).toBe(false); + abort.abort(); + expect(runSignal.aborted).toBe(true); + const response = await call; + expect(response.ok && response.result).toMatchObject({ + kind: "load", + summary: { requests: 1 }, + }); + }); + + it("ends the run when the connection drops, not on other transitions", async () => { + const started = runUntilAborted(); + const call = rpc({}); + const runSignal = await started; + fake.emitStatus("connecting"); + expect(runSignal.aborted).toBe(false); + fake.emitStatus("disconnected"); + expect(runSignal.aborted).toBe(true); + expect((await call).ok).toBe(true); + }); + + it("refuses a connection that is already down", async () => { + useClient("error"); + loadMock.impl = async () => SUMMARY; + const response = await rpc({}); + expect(response.ok).toBe(false); + expect(!response.ok && response.error.code).toBe("connection_stale"); + }); + + it("refuses --load on any method but tools/call", async () => { + const response = await rpc({ method: "tools/list" }); + expect(response.ok).toBe(false); + expect(!response.ok && response.error.code).toBe("invalid_params"); + }); + + it("leaves an ordinary tools/call on runMethod", async () => { + const response = await rpc({ load: undefined }); + expect(response.ok && response.result).toEqual({ + kind: "result", + result: {}, + appInfo: undefined, + }); + expect(runMethodMock.calls).toBe(1); + }); +}); diff --git a/clients/mcpdo/__tests__/load.test.ts b/clients/mcpdo/__tests__/load.test.ts new file mode 100644 index 000000000..0b5465189 --- /dev/null +++ b/clients/mcpdo/__tests__/load.test.ts @@ -0,0 +1,282 @@ +import { describe, it, expect, afterEach } from "vitest"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { runMcp } from "./helpers/mcp-runner.js"; +import { + createSampleTestConfig, + deleteConfigFile, +} from "../../cli/__tests__/helpers/fixtures.js"; +import { expectCliSuccess } from "../../cli/__tests__/helpers/assertions.js"; +import { callDaemon } from "../src/daemon/client.js"; +import { resolveLoadFlags } from "../src/connection/load-flags.js"; +import { formatLoadSummaryHuman } from "../src/connection/format-human.js"; +import { writeRpcOutcome } from "../src/connection/dispatch.js"; +import { + DEFAULT_LOAD_CONCURRENCY, + DEFAULT_LOAD_DURATION_MS, + type LoadSummary, +} from "@inspector/core/cli/handlers/load-test.js"; +import { PLAIN } from "@inspector/core/cli/style.js"; + +/** + * `mcpdo tools/call --load` (#2744): the flag rules, the summary rendering, + * and one end-to-end run through an auto-spawned daemon. + */ + +const SUMMARY: LoadSummary = { + tool: "echo", + concurrency: 4, + durationMs: 1500, + requestLimit: 40, + rate: 20, + requests: 30, + ok: 27, + toolErrors: 1, + errors: 2, + errorRate: 0.1, + elapsedMs: 1500, + throughput: 20, + latencyMs: { min: 1, mean: 2.5, p50: 2, p90: 4, p99: 5, max: 6 }, + errorMessages: [{ message: "Request timed out", count: 2 }], +}; + +describe("resolveLoadFlags", () => { + it("returns undefined for an ordinary call", () => { + expect(resolveLoadFlags({ toolName: "echo" })).toBeUndefined(); + }); + + it("applies the engine's defaults", () => { + expect(resolveLoadFlags({ load: true, toolName: "echo" })).toEqual({ + concurrency: DEFAULT_LOAD_CONCURRENCY, + durationMs: DEFAULT_LOAD_DURATION_MS, + }); + }); + + it("passes every bound through", () => { + expect( + resolveLoadFlags({ + load: true, + toolName: "echo", + loadConcurrency: 3, + loadDuration: 2, + loadRequests: 9, + loadRate: 5, + }), + ).toEqual({ concurrency: 3, durationMs: 2000, requests: 9, rate: 5 }); + }); + + it.each([ + [{ loadConcurrency: 2 }, "--load-concurrency requires --load."], + [{ loadDuration: 1 }, "--load-duration requires --load."], + [{ loadRequests: 1 }, "--load-requests requires --load."], + [{ loadRate: 1 }, "--load-rate requires --load."], + [ + { load: true, appInfo: true }, + "--load cannot be combined with --app-info.", + ], + ])("rejects %j", (flags, message) => { + expect(() => resolveLoadFlags({ toolName: "echo", ...flags })).toThrow( + message, + ); + }); + + it("requires a tool name", () => { + expect(() => resolveLoadFlags({ load: true })).toThrow( + "--load requires a tool name.", + ); + }); +}); + +describe("formatLoadSummaryHuman", () => { + it("renders bounds, outcomes, latency and top errors", () => { + const text = formatLoadSummaryHuman(SUMMARY, PLAIN); + expect(text).toBe( + [ + "Load test `echo` (concurrency 4, duration 1.5s, limit 40, rate 20/s)", + " Requests: 30 in 1.50s (20/s)", + " Outcomes: 27 ok, 1 tool errors, 2 errors", + " Error rate: 10.00%", + " Latency: min 1 / mean 2.5 / p50 2 / p90 4 / p99 5 / max 6 ms", + " Top errors:", + " 2× Request timed out", + ].join("\n"), + ); + }); + + it("handles a run where nothing completed", () => { + const text = formatLoadSummaryHuman({ + concurrency: 1, + requests: 0, + ok: 0, + toolErrors: 0, + errors: 0, + errorRate: 0, + elapsedMs: 0, + throughput: 0, + latencyMs: null, + errorMessages: [], + }); + expect(text).toContain("Load test (concurrency 1)"); + expect(text).toContain("Error rate: 0.00%"); + expect(text).toContain("Latency: no calls completed"); + expect(text).not.toContain("Top errors"); + }); +}); + +describe("writeRpcOutcome for a load result", () => { + async function capture(format: "text" | "json"): Promise { + let out = ""; + const original = process.stdout.write; + process.stdout.write = ((chunk: unknown, ...rest: unknown[]) => { + out += String(chunk); + const cb = rest.find((r) => typeof r === "function") as + | (() => void) + | undefined; + cb?.(); + return true; + }) as typeof process.stdout.write; + try { + await writeRpcOutcome({ format, style: PLAIN }, "tools/call", "echo", { + kind: "load", + summary: SUMMARY, + }); + } finally { + process.stdout.write = original; + } + return out; + } + + it("prints the bare summary as JSON", async () => { + expect(JSON.parse(await capture("json"))).toEqual(SUMMARY); + }); + + it("prints the human summary as text", async () => { + expect(await capture("text")).toContain("Load test `echo`"); + }); +}); + +/** Deadline for one `runMcp` call that reaches the auto-spawned daemon. */ +const DAEMON_COMMAND_MS = 20_000; +/** + * The longest test here runs three daemon commands in sequence, so its own + * budget has to exceed all three deadlines (AGENTS.md: an inner budget is + * strictly smaller than the one enclosing it). Otherwise a stalled command + * outlives its test and overlaps teardown while `runMcp` still holds the + * captured stdout/stderr and the env overrides. + */ +const DAEMON_TEST_MS = 3 * DAEMON_COMMAND_MS + 5_000; + +describe("mcpdo tools/call --load", { timeout: DAEMON_TEST_MS }, () => { + let configPath: string | undefined; + let storageDir: string | undefined; + + afterEach(async () => { + if (storageDir) { + const socketPath = path.join(storageDir, "mcpdod.sock"); + if (fs.existsSync(socketPath)) { + try { + await callDaemon("daemon/stop", {}, { socketPath, timeoutMs: 2000 }); + } catch { + // already stopped + } + } + fs.rmSync(storageDir, { recursive: true, force: true }); + storageDir = undefined; + } + if (configPath) { + deleteConfigFile(configPath); + configPath = undefined; + } + }); + + function env(): Record { + storageDir = fs.mkdtempSync(path.join(os.tmpdir(), "mcp-load-e2e-")); + return { + MCP_STORAGE_DIR: storageDir, + MCP_INSPECTOR_DAEMON_DIR: storageDir, + MCP_ALLOW_DEFAULT_CONNECTION: "1", + }; + } + + it("loads a tool over a live connection and leaves it usable", async () => { + configPath = createSampleTestConfig(); + const e = env(); + expectCliSuccess( + await runMcp(["connect", "test-stdio", "--config", configPath], { + env: e, + timeout: DAEMON_COMMAND_MS, + }), + ); + + const loaded = await runMcp( + [ + "tools/call", + "echo", + "message:=hi", + "--load", + "--load-requests", + "12", + "--load-concurrency", + "3", + "--format", + "json", + ], + { env: e, timeout: DAEMON_COMMAND_MS }, + ); + expectCliSuccess(loaded); + expect(JSON.parse(loaded.stdout)).toMatchObject({ + tool: "echo", + concurrency: 3, + requestLimit: 12, + requests: 12, + ok: 12, + errors: 0, + errorRate: 0, + }); + + // The queue slot is released: an ordinary call runs afterwards. + expectCliSuccess( + await runMcp( + ["tools/call", "echo", "message:=after", "--format", "json"], + { + env: e, + timeout: DAEMON_COMMAND_MS, + }, + ), + ); + }); + + it("fails before loading when the tool does not exist", async () => { + configPath = createSampleTestConfig(); + const e = env(); + expectCliSuccess( + await runMcp(["connect", "test-stdio", "--config", configPath], { + env: e, + timeout: DAEMON_COMMAND_MS, + }), + ); + const result = await runMcp( + ["tools/call", "no_such_tool", "--load", "--load-requests", "2"], + { env: e, timeout: DAEMON_COMMAND_MS }, + ); + expect(result.exitCode).toBe(5); + expect(result.stderr).toContain("Tool 'no_such_tool' not found on server."); + }); + + it.each([ + [ + ["tools/call", "echo", "--load-requests", "2"], + "--load-requests requires --load.", + ], + [["tools/call", "--load"], "--load requires a tool name."], + [ + ["tools/call", "echo", "--load", "--load-concurrency", "0"], + "--load-concurrency must be a positive integer.", + ], + ])("rejects %j before reaching a daemon", async (args, message) => { + const result = await runMcp(args, { env: env() }); + expect(result.exitCode).not.toBe(0); + expect(result.stderr).toContain(message); + }); +}); diff --git a/clients/mcpdo/src/connection/dispatch.ts b/clients/mcpdo/src/connection/dispatch.ts index 1d4d92c02..1246f02c1 100644 --- a/clients/mcpdo/src/connection/dispatch.ts +++ b/clients/mcpdo/src/connection/dispatch.ts @@ -5,6 +5,7 @@ import type { MethodArgs, } from "@inspector/core/cli/handlers/method-types.js"; import type { OutputFormat } from "@inspector/core/cli/handlers/format-output.js"; +import type { LoadOptions } from "@inspector/core/cli/handlers/load-test.js"; import { writeConnectionOutput } from "./format-connection.js"; import { styleFromOpts, type Style } from "@inspector/core/cli/style.js"; import { promptElicitation } from "./elicitation-prompt.js"; @@ -24,6 +25,8 @@ export type ConnectionDispatchOpts = { plain?: boolean; connection?: string; requireExplicit: boolean; + /** `tools/call --load` (#2744): run the load engine daemon-side. */ + load?: LoadOptions; }; /** @@ -45,6 +48,7 @@ export async function dispatchConnectionRpc( method, name: stripAt(opts.connection), requireExplicit: opts.requireExplicit, + ...(opts.load !== undefined && { load: opts.load }), }; const { socketPath } = await ensureDaemon(); @@ -150,6 +154,13 @@ export async function writeRpcOutcome( outcome: RpcResult, ): Promise { const { format, style } = out; + if (outcome.kind === "load") { + await writeConnectionOutput( + { format, style }, + { kind: "load", summary: outcome.summary }, + ); + return; + } if (outcome.kind === "elicitation-pending") { await writeConnectionOutput( { format, style }, diff --git a/clients/mcpdo/src/connection/format-connection.ts b/clients/mcpdo/src/connection/format-connection.ts index ed5077d1e..4944a6e81 100644 --- a/clients/mcpdo/src/connection/format-connection.ts +++ b/clients/mcpdo/src/connection/format-connection.ts @@ -12,6 +12,7 @@ import { } from "@inspector/core/cli/error-handler.js"; import type { OutputFormat } from "@inspector/core/cli/handlers/format-output.js"; import type { CliAppInfo } from "@inspector/core/cli/handlers/method-types.js"; +import type { LoadSummary } from "@inspector/core/cli/handlers/load-test.js"; import { formatAppInfoHuman, formatAppInfoListHuman, @@ -23,6 +24,7 @@ import { formatConnectionInfoHuman, formatConnectionsListHuman, formatElicitationPendingHuman, + formatLoadSummaryHuman, formatSkillVerifyListHuman, formatStreamEventHuman, } from "./format-human.js"; @@ -76,6 +78,7 @@ export type ConnectionWriteKind = exitCode?: number; } | { kind: "stream-event"; data: unknown } + | { kind: "load"; summary: LoadSummary } | { kind: "servers/list"; servers: unknown[]; @@ -227,6 +230,8 @@ function jsonPayload(payload: ConnectionWriteKind): unknown { return payload.lines; case "stream-event": return payload.data; + case "load": + return payload.summary; case "servers/list": return { servers: payload.servers, @@ -342,6 +347,8 @@ function humanPayload( : formatAppInfoListHuman(payload.lines, style); case "stream-event": return formatStreamEventHuman(payload.data, style); + case "load": + return formatLoadSummaryHuman(payload.summary, style); case "servers/list": return formatServersListHuman(payload.servers, style, payload.source); case "servers/show": diff --git a/clients/mcpdo/src/connection/format-human.ts b/clients/mcpdo/src/connection/format-human.ts index 966c48b95..28650d35f 100644 --- a/clients/mcpdo/src/connection/format-human.ts +++ b/clients/mcpdo/src/connection/format-human.ts @@ -7,6 +7,7 @@ import { PLAIN, type Style } from "@inspector/core/cli/style.js"; import { isSafeLinkTarget } from "./sanitize.js"; import { parseFormSchema } from "./form-schema.js"; import type { ElicitationPendingInfo } from "../daemon/protocol.js"; +import type { LoadSummary } from "@inspector/core/cli/handlers/load-test.js"; type JsonObject = Record; @@ -974,6 +975,50 @@ export function formatStreamEventHuman( return JSON.stringify(ev, null, 2); } +/** + * `tools/call --load` summary (#2744): the run's shape, its outcome counts, + * the latency distribution, and the most frequent thrown messages. Every + * figure is also in `--format json`, which is the form to script against. + */ +export function formatLoadSummaryHuman( + summary: LoadSummary, + style: Style = PLAIN, +): string { + const tool = + summary.tool !== undefined ? ` ${code(style, summary.tool)}` : ""; + const bounds = [ + `concurrency ${summary.concurrency}`, + ...(summary.durationMs !== undefined + ? [`duration ${summary.durationMs / 1000}s`] + : []), + ...(summary.requestLimit !== undefined + ? [`limit ${summary.requestLimit}`] + : []), + ...(summary.rate !== undefined ? [`rate ${summary.rate}/s`] : []), + ]; + const failed = summary.toolErrors + summary.errors; + const rate = `${(summary.errorRate * 100).toFixed(2)}%`; + const lines = [ + `${heading(style, "Load test")}${tool} ${style.dim(`(${bounds.join(", ")})`)}`, + ` Requests: ${summary.requests} in ${(summary.elapsedMs / 1000).toFixed(2)}s (${summary.throughput}/s)`, + ` Outcomes: ${style.green(`${summary.ok} ok`)}, ${summary.toolErrors} tool errors, ${summary.errors} errors`, + ` Error rate: ${failed > 0 ? style.red(rate) : rate}`, + ]; + const l = summary.latencyMs; + lines.push( + l === null + ? ` Latency: ${style.dim("no calls completed")}` + : ` Latency: min ${l.min} / mean ${l.mean} / p50 ${l.p50} / p90 ${l.p90} / p99 ${l.p99} / max ${l.max} ms`, + ); + if (summary.errorMessages.length > 0) { + lines.push(` ${style.bold("Top errors:")}`); + for (const { message, count } of summary.errorMessages) { + lines.push(` ${count}× ${message}`); + } + } + return lines.join("\n"); +} + /** * Dispatch human formatting for an RPC method result. * Returns null when the caller should fall back to pretty JSON. diff --git a/clients/mcpdo/src/connection/load-flags.ts b/clients/mcpdo/src/connection/load-flags.ts new file mode 100644 index 000000000..79c4b24d2 --- /dev/null +++ b/clients/mcpdo/src/connection/load-flags.ts @@ -0,0 +1,62 @@ +import { + CliExitCodeError, + EXIT_CODES, +} from "@inspector/core/cli/error-handler.js"; +import { + resolveLoadOptions, + type LoadOptions, +} from "@inspector/core/cli/handlers/load-test.js"; + +/** + * `tools/call --load` (#2744): turn mcpdo's flag values into the shared + * engine's {@link LoadOptions}, refusing the same combinations the one-shot + * CLI refuses (#2439). Each `--load-*` bound is meaningless without `--load`, + * and `--app-info` shapes a single result where a load run prints a summary, + * so both are usage errors rather than silently ignored. Value parsing itself + * (positive integers, positive numbers) is done by commander through the + * engine's own parsers, so the two CLIs reject the same inputs. + * + * Returns `undefined` when `--load` is not set: an ordinary `tools/call`. + */ +export function resolveLoadFlags(flags: { + load?: boolean; + loadConcurrency?: number; + loadDuration?: number; + loadRequests?: number; + loadRate?: number; + appInfo?: boolean; + toolName?: string; +}): LoadOptions | undefined { + const bounds = [ + ["--load-concurrency", flags.loadConcurrency], + ["--load-duration", flags.loadDuration], + ["--load-requests", flags.loadRequests], + ["--load-rate", flags.loadRate], + ] as const; + if (flags.load !== true) { + for (const [flag, value] of bounds) { + if (value !== undefined) { + throw usage(`${flag} requires --load.`); + } + } + return undefined; + } + if (flags.appInfo === true) { + throw usage("--load cannot be combined with --app-info."); + } + // Checked here rather than left to the daemon so a missing name fails + // before a connection is resolved, with the flag a user can act on. + if (!flags.toolName) { + throw usage("--load requires a tool name.", "missing_tool_name"); + } + return resolveLoadOptions({ + concurrency: flags.loadConcurrency, + durationSec: flags.loadDuration, + requests: flags.loadRequests, + rate: flags.loadRate, + }); +} + +function usage(message: string, code = "load_flag_conflict"): CliExitCodeError { + return new CliExitCodeError(EXIT_CODES.USAGE, message, { code }); +} diff --git a/clients/mcpdo/src/connection/mcp.ts b/clients/mcpdo/src/connection/mcp.ts index 0d1c10a9d..b802d2144 100644 --- a/clients/mcpdo/src/connection/mcp.ts +++ b/clients/mcpdo/src/connection/mcp.ts @@ -55,6 +55,12 @@ import { CONNECTION_RPC_METHODS, type MethodArgs, } from "@inspector/core/cli/handlers/method-types.js"; +import { + positiveIntegerParser, + positiveNumberParser, + type LoadOptions, +} from "@inspector/core/cli/handlers/load-test.js"; +import { resolveLoadFlags } from "./load-flags.js"; import { authorizeInFrontend } from "./authorize.js"; import { AUTH_HELPER_COMMAND, @@ -1342,7 +1348,33 @@ function registerRpcCommands(program: CommandType): void { "--task", "Request a task-augmented tool call (legacy servers; on modern the server decides)", ) - .option("--app-info", "Probe MCP App metadata only"); + .option("--app-info", "Probe MCP App metadata only") + // `--load` (#2744): the one-shot CLI's load mode, run inside the + // daemon against this connection. Same flags, same summary. + .option( + "--load", + "Load-test the tool: call it repeatedly and concurrently, then print a latency and error-rate summary instead of a result. Bounded by --load-duration (default 10s) and/or --load-requests. Later commands on this connection wait until the run ends.", + ) + .option( + "--load-concurrency ", + "With --load: most calls in flight at once (positive integer; default 10).", + positiveIntegerParser("--load-concurrency"), + ) + .option( + "--load-duration ", + "With --load: stop starting calls after this many seconds; in-flight calls still finish (default 10 when --load-requests is not set).", + positiveNumberParser("--load-duration"), + ) + .option( + "--load-requests ", + "With --load: stop after starting this many calls (positive integer).", + positiveIntegerParser("--load-requests"), + ) + .option( + "--load-rate ", + "With --load: start at most this many calls per second, spread evenly (positive number; default unlimited).", + positiveNumberParser("--load-rate"), + ); cmd.action( async ( toolNamePos: string | undefined, @@ -1356,14 +1388,20 @@ function registerRpcCommands(program: CommandType): void { toolArgFlag: (o.toolArg ?? {}) as Record, toolArgsJson: o.toolArgsJson as string | undefined, }); - await runRpc(program, method, { - toolName, - toolArg, - toolMeta: o.toolMetadata, - metadata: o.metadata, - task: o.task === true, - appInfo: o.appInfo === true, - }); + const load = resolveLoadFlags({ ...o, toolName }); + await runRpc( + program, + method, + { + toolName, + toolArg, + toolMeta: o.toolMetadata, + metadata: o.metadata, + task: o.task === true, + appInfo: o.appInfo === true, + }, + load, + ); }, ); break; @@ -1613,6 +1651,7 @@ async function runRpc( program: CommandType, method: string, methodArgs: MethodArgs, + load?: LoadOptions, ): Promise { const opts = program.opts(); await dispatchConnectionRpc(method, methodArgs, { @@ -1620,6 +1659,7 @@ async function runRpc( plain: opts.plain === true, connection: opts.connection, requireExplicit: requireExplicitConnection(), + load, }); } diff --git a/clients/mcpdo/src/daemon/protocol.ts b/clients/mcpdo/src/daemon/protocol.ts index e6174eeff..1ecfb2065 100644 --- a/clients/mcpdo/src/daemon/protocol.ts +++ b/clients/mcpdo/src/daemon/protocol.ts @@ -8,6 +8,10 @@ import type { MethodArgs, } from "@inspector/core/cli/handlers/method-types.js"; import type { ErrorEnvelope } from "@inspector/core/cli/error-handler.js"; +import type { + LoadOptions, + LoadSummary, +} from "@inspector/core/cli/handlers/load-test.js"; import type { Implementation, ProtocolEra, @@ -89,6 +93,14 @@ export type RpcParams = ConnectionNameParams & * only ever a daemon-internal caller. */ interactive?: boolean; + /** + * `tools/call --load` (#2744): run the shared load engine against this + * connection in place of one call, and return its {@link LoadSummary} as + * a `kind: "load"` result. The whole run is one `rpc` op, so it holds + * this connection's rpc queue until it finishes — later commands on the + * same connection wait for it; other connections are unaffected. + */ + load?: LoadOptions; }; export type DaemonRequest = { @@ -363,6 +375,11 @@ export type RpcResult = /** Non-zero when the emitted report is itself a failure (`--verify`). */ exitCode?: number; } + | { + /** `tools/call --load`: the summary of the whole run. */ + kind: "load"; + summary: LoadSummary; + } | { /** * The call surfaced an elicitation while diff --git a/clients/mcpdo/src/daemon/server.ts b/clients/mcpdo/src/daemon/server.ts index 91f3e41cc..ea866861c 100644 --- a/clients/mcpdo/src/daemon/server.ts +++ b/clients/mcpdo/src/daemon/server.ts @@ -12,6 +12,10 @@ import { } from "@inspector/core/cli/handlers/run-method.js"; import { MCPDO_SKILL_BUDGET_HINT } from "@inspector/core/cli/handlers/skills-verify.js"; import type { MethodArgs } from "@inspector/core/cli/handlers/method-types.js"; +import { + runToolLoad, + type LoadOptions, +} from "@inspector/core/cli/handlers/load-test.js"; import { acceptDaemonConnection, removeStaleDaemonSocket, @@ -598,7 +602,16 @@ export class DaemonServer { // raw `params.interactive` (boolean | undefined) is threaded to // `liveClientFor`, and `authRequiredError` treats only `=== true` as a // human — so absent falls to the agent-tuned wording. - const park = params.interactive === false; + if (params.load !== undefined && params.method !== "tools/call") { + throw new CliExitCodeError( + EXIT_CODES.USAGE, + "--load requires tools/call", + { code: "invalid_params" }, + ); + } + // A load run never parks: its elicitations are declined, not relayed + // (see runLoadOnClient). + const park = params.interactive === false && params.load === undefined; // Parking needs the connection *name* for the pending payload and for // teardown-keyed cancellation; resolve it before reviving the client. const connectionName = park @@ -620,6 +633,15 @@ export class DaemonServer { { code: "caller_gone" }, ); } + if (params.load !== undefined) { + return this.runLoadOnClient( + client, + requestId, + params, + params.load, + signal, + ); + } return park ? this.runRpcParked(client, connectionName!, requestId, params) : this.runRpcOnClient(client, requestId, params, elicitation, signal); @@ -687,6 +709,65 @@ export class DaemonServer { return toRpcResult(outcome, params.method); } + /** + * `tools/call --load` (#2744): the shared load engine, run here against the + * daemon's live connection so the front-end makes one round trip for the + * whole run instead of one per call. It runs inside the per-client rpc + * queue like any other op, so later commands on this connection wait for + * it to finish. + */ + private async runLoadOnClient( + client: InspectorClient, + requestId: string, + params: RpcParams, + load: LoadOptions, + signal?: AbortSignal, + ): Promise { + const methodArgs = stripConnectionFields(params); + this.assertNoParkedCall(client); + if (isTerminalStatus(client.getStatus())) { + throw new CliExitCodeError( + EXIT_CODES.UNREACHABLE, + "The connection dropped before this command could run; re-run the command to reconnect.", + { code: "connection_stale" }, + ); + } + // Many calls share one bridge subscriber, so an elicitation could not be + // routed back to the call that raised it, and nobody is at the stream to + // answer one mid-run anyway: cancel each, as the in-process default does. + const unwire = wireElicitationBridge( + client, + autoCancelElicitationChannel, + requestId, + ); + // Ends the run early. The caller hanging up is one reason; the connection + // dropping is the other — every further call would fail at once, turning + // the rest of the window into a loop of instant failures that holds the + // queue for nothing. The summary still reports what completed. + const stop = new AbortController(); + const onCallerGone = () => stop.abort(); + signal?.addEventListener("abort", onCallerGone, { once: true }); + const onStatus = ( + event: TypedEventGeneric, + ) => { + if (isTerminalStatus(event.detail)) stop.abort(); + }; + client.addEventListener("statusChange", onStatus); + // In-flight calls abort with the run, not only new ones (#1783). + const clearAmbientSignal = client.setAmbientRequestSignal(stop.signal); + try { + const summary = await runToolLoad(client, methodArgs, load, { + signal: stop.signal, + }); + return { kind: "load", summary }; + } finally { + clearAmbientSignal(); + client.removeEventListener("statusChange", onStatus); + signal?.removeEventListener("abort", onCallerGone); + unwire(); + } + } + /** * A connection with a parked call must not accept new rpcs: the bridge * routes elicitations to its oldest subscriber, so a second in-flight call @@ -1106,13 +1187,14 @@ function stripConnectionFields( // runMethod's `format === "json"` branch collect app info (an extra // resources/read) whose result the frontend discards. `interactive` is // daemon routing (elicitation parking, message audience), not a method - // argument. - const { name, requireExplicit, format, interactive, method, ...rest } = + // argument. `load` is consumed by runLoadOnClient, never by runMethod. + const { name, requireExplicit, format, interactive, load, method, ...rest } = params; void name; void requireExplicit; void format; void interactive; + void load; return { method, ...rest }; } diff --git a/clients/web/src/test/integration/mcp/inspectorClient-ambient-signal.test.ts b/clients/web/src/test/integration/mcp/inspectorClient-ambient-signal.test.ts index c67166a05..eb072e97a 100644 --- a/clients/web/src/test/integration/mcp/inspectorClient-ambient-signal.test.ts +++ b/clients/web/src/test/integration/mcp/inspectorClient-ambient-signal.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, afterEach } from "vitest"; import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js"; import { createTransportNode } from "@inspector/core/mcp/node/transport.js"; +import { eraToVersionNegotiation } from "@inspector/core/mcp/types.js"; import { createTestServerHttp, createTestServerInfo, @@ -130,6 +131,61 @@ describe("ambient request signal cancels non-tool requests (#1783)", () => { dispose(); }); + // A modern-era tool call goes through the task session, not + // `getRequestOptions`, so the ambient signal has to be folded in on that + // path separately (#2744: `mcpdo tools/call --load` relies on it to cancel + // in-flight calls when its caller hangs up). + it("aborts an in-flight modern-era tool call when the ambient signal fires", async () => { + const started = deferred(); + const aborted = deferred(); + const slowServer = createTestServerHttp({ + serverInfo: createTestServerInfo("ambient-signal-modern", "1.0.0"), + tools: [ + { + name: "slow_tool", + description: "Runs until the client cancels it", + handler: (_params, _context, extra) => { + extra?.signal?.addEventListener( + "abort", + () => aborted.resolve(true), + { once: true }, + ); + started.resolve(); + // Never settles on its own; only the abort ends the call. + return new Promise(() => {}); + }, + }, + ], + modern: {}, + }); + await slowServer.start(); + server = slowServer; + + const connected = new InspectorClient( + { type: "streamable-http", url: slowServer.url }, + { + environment: { transport: createTransportNode }, + versionNegotiation: eraToVersionNegotiation("modern"), + }, + ); + await connected.connect(); + client = connected; + const { tools } = await connected.listTools(); + const tool = tools.find((t) => t.name === "slow_tool"); + + const controller = new AbortController(); + const dispose = connected.setAmbientRequestSignal(controller.signal); + const call = connected.callTool(tool!, {}); + const rejected = expect(call).rejects.toThrow(); + + await started.promise; + controller.abort(); + + await rejected; + expect(await aborted.promise).toBe(true); + dispose(); + }); + it("leaves a normal request untouched when the ambient signal never fires", async () => { // A control: setting an ambient signal that is never aborted must not break // an ordinary request. Uses a template that answers immediately. diff --git a/core/cli/handlers/load-test.ts b/core/cli/handlers/load-test.ts index c8b3f4d9a..520f12a85 100644 --- a/core/cli/handlers/load-test.ts +++ b/core/cli/handlers/load-test.ts @@ -9,9 +9,8 @@ import type { MethodArgs } from "./method-types.js"; * traffic over the connection the Inspector already set up — transport, OAuth * and headers included — and report latency and error rate at the end. * - * Lives in `core/cli/` rather than the CLI client so `mcpdo` can reuse it - * against a daemon connection without a second copy; only the one-shot CLI - * exposes it today. + * Lives in `core/cli/` so the one-shot CLI and `mcpdo` share one engine: + * `mcpdo` runs it inside the daemon against a live connection (#2744). * * By default all calls share ONE connection, so this measures how a server * handles concurrent requests on a session. `connections` (#2746) spreads the @@ -176,17 +175,66 @@ export function resolveLoadOptions(flags: { export const MAX_ERROR_KINDS = 10; export const OTHER_ERRORS = "(other)"; +/** + * Slots claimed between forced yields to the event loop. Without + * `--load-rate` nothing in the claim path waits on a timer, so calls that + * settle without I/O (a client that rejects up front) would keep the run in + * one unbroken microtask chain: no timer, socket close or other request gets + * a turn until the run's bounds expire. Inside the mcpdo daemon that blocks + * every connection, and the caller's own hang-up. + */ +export const YIELD_EVERY_SLOTS = 64; + +const yieldToEventLoop = (): Promise => + new Promise((resolve) => setImmediate(resolve)); + /** Injected so tests can drive the clock instead of waiting on it. */ export type LoadClock = { now: () => number; - sleep: (ms: number) => Promise; + /** A clock that can, should drop its timer when `signal` aborts. */ + sleep: (ms: number, signal?: AbortSignal) => Promise; }; const realClock: LoadClock = { now: () => performance.now(), - sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), + sleep: (ms, signal) => + new Promise((resolve) => { + // Cleared on abort, not left to fire: a `--load-rate 0.001` slot is a + // 1000s timer, and a pending one keeps the process (the mcpdo daemon + // during shutdown) alive until it fires. + const timer = setTimeout(() => { + signal?.removeEventListener("abort", cancel); + resolve(); + }, ms); + const cancel = (): void => { + clearTimeout(timer); + resolve(); + }; + signal?.addEventListener("abort", cancel, { once: true }); + }), }; +/** + * {@link LoadClock.sleep}, cut short when `signal` aborts, whether or not the + * clock honors the signal itself, so a slow `--load-rate` slot never holds a + * cancelled run open for a whole interval. + */ +function sleepUnlessAborted( + clock: LoadClock, + ms: number, + signal: AbortSignal | undefined, +): Promise { + if (!signal) return clock.sleep(ms); + return new Promise((resolve) => { + const done = (): void => { + signal.removeEventListener("abort", done); + resolve(); + }; + signal.addEventListener("abort", done, { once: true }); + void clock.sleep(ms, signal).then(done); // `sleep` never rejects + }); +} + /** Nearest-rank percentile over an ascending-sorted, non-empty array. */ export function percentile(sorted: number[], p: number): number { const rank = Math.ceil((p / 100) * sorted.length); @@ -253,11 +301,15 @@ class ErrorTally { * owed, so a stall — one slow call at concurrency 1, or timers firing late — * is never followed by a catch-up burst against a server that may just be * recovering. A slot past the duration is never started. + * + * Aborting `signal` ends the run early: no further call starts, in-flight + * calls are left to settle, and the summary covers what completed. */ export async function runLoad( call: (worker: number) => Promise, options: LoadOptions, clock: LoadClock = realClock, + signal?: AbortSignal, ): Promise { const { concurrency, durationMs, requests, rate } = options; const latencies: number[] = []; @@ -293,7 +345,13 @@ export async function runLoad( turnstile = new Promise((resolve) => (release = resolve)); await ahead; try { + // Yield while holding the turnstile: every other worker is queued + // behind it, so the whole run pauses, not just this worker. + if (started > 0 && started % YIELD_EVERY_SLOTS === 0) { + await yieldToEventLoop(); + } for (;;) { + if (signal?.aborted) return false; if (requests !== undefined && started >= requests) return false; const now = clock.now(); if (now >= deadline) return false; @@ -309,7 +367,7 @@ export async function runLoad( now - nextStart < interval ? nextStart + interval : now + interval; return true; } - await clock.sleep(scheduled - now); + await sleepUnlessAborted(clock, scheduled - now, signal); } } finally { release(); @@ -318,6 +376,9 @@ export async function runLoad( const worker = async (index: number): Promise => { while (await claimSlot()) { + // Another worker's call can abort the run between this worker taking + // its slot and resuming here. + if (signal?.aborted) return; const t0 = clock.now(); try { const outcome = await call(index); @@ -381,10 +442,11 @@ export async function runLoadAcrossConnections( callOn: (connection: C) => Promise, options: LoadOptions, clock: LoadClock = realClock, + signal?: AbortSignal, ): Promise { const requested = options.connections ?? DEFAULT_LOAD_CONNECTIONS; if (requested <= 1) { - return runLoad(() => callOn(primary), options, clock); + return runLoad(() => callOn(primary), options, clock, signal); } if (!openConnection) { throw new Error( @@ -417,6 +479,7 @@ export async function runLoadAcrossConnections( (worker) => callOn(pool[worker % pool.length]!), options, clock, + signal, ); return { ...summary, @@ -451,6 +514,8 @@ export async function runToolLoad( extra: { openConnection?: () => Promise; clock?: LoadClock; + /** Ends the run early; see {@link runLoad}. */ + signal?: AbortSignal; } = {}, ): Promise { if (!args.toolName) { @@ -499,6 +564,7 @@ export async function runToolLoad( call, options, extra.clock, + extra.signal, ); return { tool: tool.name, ...summary }; } diff --git a/core/mcp/inspectorClient.ts b/core/mcp/inspectorClient.ts index a202a748f..26b28af57 100644 --- a/core/mcp/inspectorClient.ts +++ b/core/mcp/inspectorClient.ts @@ -4054,6 +4054,11 @@ export class InspectorClient extends InspectorClientEventTarget { ): Promise { const session = this.taskSession; if (!session) throw new Error("Client is not connected"); + // This path calls the task session directly rather than through + // `getRequestOptions`, so it folds in the ambient request signal itself + // (#1783): a modern-era or task-required call must abort with its caller + // too, or a daemon caller's hang-up leaves it holding the connection. + signal = combineAbortSignals(signal, this.ambientRequestSignal); const headers = this.mirroredTaskParamHeaders(tool, args); let lastTask: InspectorTask | undefined; let outcomeEmitted = false;