diff --git a/clients/web/src/test/core/mcp/authResyncPropagation.test.ts b/clients/web/src/test/core/mcp/authResyncPropagation.test.ts index 96e03fc3..604ac63b 100644 --- a/clients/web/src/test/core/mcp/authResyncPropagation.test.ts +++ b/clients/web/src/test/core/mcp/authResyncPropagation.test.ts @@ -117,7 +117,7 @@ describe("RemoteClientTransport send contract", () => { const started = Date.now(); await expect( transport.send({ jsonrpc: "2.0", id: 1, method: "tools/list" }), - ).rejects.toThrow(/Timed out waiting for MCP response on SSE/i); + ).rejects.toThrow(/Request timed out/); expect(Date.now() - started).toBeLessThan(5000); await transport.close(); diff --git a/clients/web/src/test/core/mcp/fetchTracking.test.ts b/clients/web/src/test/core/mcp/fetchTracking.test.ts index 61dc817e..3d834caa 100644 --- a/clients/web/src/test/core/mcp/fetchTracking.test.ts +++ b/clients/web/src/test/core/mcp/fetchTracking.test.ts @@ -826,6 +826,56 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { expect(last.closedAt).toBeInstanceOf(Date); }); + it("accepts bare CR line endings, and a CRLF split across chunks", async () => { + const server = controlledStream(); + const { updates } = await trackStream(server.stream); + server.push("data: a\r\rdata: b\r"); + await flush(); + // The trailing CR is held: it may be the first half of a CRLF. + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1]); + server.push("\n\r\n"); + await flush(); + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]); + }); + + it("classifies the content type and the method case-insensitively", async () => { + const server = controlledStream(); + const updates: unknown[] = []; + const fetcher = createFetchTracker( + (async () => + new Response(server.stream, { + headers: { "content-type": "Text/Event-Stream" }, + })) as typeof fetch, + { updateStream: (id, state) => updates.push({ id, state }) }, + ); + await fetcher("https://example.com/mcp", { method: "get" }); + expect(updates).toHaveLength(1); + }); + + it("takes the method from a Request input when init carries none", async () => { + // A bounded POST reply must go down the body-capture path, not the + // watcher, when the caller passed a `Request` rather than `init`. + const bodies: string[] = []; + const updates: unknown[] = []; + const tracked: FetchRequestEntryBase[] = []; + const fetcher = createFetchTracker( + (async () => + new Response("data: x\n\n", { + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch, + { + trackRequest: (entry) => tracked.push(entry), + updateResponseBody: (_id, body) => bodies.push(body), + updateStream: (id, state) => updates.push({ id, state }), + }, + ); + await fetcher(new Request("https://example.com/mcp", { method: "POST" })); + await flush(); + expect(tracked[0]?.method).toBe("POST"); + expect(bodies).toEqual(["data: x\n\n"]); + expect(updates).toEqual([]); + }); + it("reports the close when the server ends the stream", async () => { const server = controlledStream(); const { updates, id } = await trackStream(server.stream); diff --git a/clients/web/src/test/core/mcp/inspectorClient-diagnostics-callbacks.test.ts b/clients/web/src/test/core/mcp/inspectorClient-diagnostics-callbacks.test.ts new file mode 100644 index 00000000..cae7981b --- /dev/null +++ b/clients/web/src/test/core/mcp/inspectorClient-diagnostics-callbacks.test.ts @@ -0,0 +1,75 @@ +import { describe, it, expect } from "vitest"; +import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; +import type { JSONRPCRequest } from "@modelcontextprotocol/client"; +import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js"; +import type { MessageTrackingCallbacks } from "@inspector/core/mcp/messageTrackingTransport.js"; + +/** + * The outstanding-request bookkeeping behind the connection diagnostics + * (#2318), driven through the message-tracking callbacks the client hands its + * transport — the same seam the transport uses, without a connection. + */ +describe("InspectorClient diagnostics tracking callbacks", () => { + interface Internals { + createMessageTrackingCallbacks: () => MessageTrackingCallbacks; + } + + function makeClient(): { + client: InspectorClient; + callbacks: MessageTrackingCallbacks; + } { + const client = new InspectorClient( + { type: "stdio", command: "noop", args: [] }, + // Never connects; the callbacks are exercised directly. + { environment: { transport: () => ({}) as never } }, + ); + const callbacks = ( + client as unknown as Internals + ).createMessageTrackingCallbacks(); + return { client, callbacks }; + } + + const request = (id: number): JSONRPCRequest => ({ + jsonrpc: "2.0", + id, + method: "tools/list", + }); + + const outstandingIds = (client: InspectorClient) => + client.getConnectionDiagnostics().outstandingRequests.map((r) => r.id); + + it("rolls a request back when its send failed before reaching the wire", () => { + const { client, callbacks } = makeClient(); + callbacks.trackRequest!(request(1), "client"); + expect(outstandingIds(client)).toEqual([1]); + callbacks.trackSendFailure!(request(1), new Error("connection closed")); + expect(outstandingIds(client)).toEqual([]); + }); + + it("keeps a request whose send failed with a request timeout", () => { + // The browser's remote transport awaits the response inside `send`, so + // its relay timeout surfaces as a send failure — the request reached the + // server and is still unanswered. + const { client, callbacks } = makeClient(); + callbacks.trackRequest!(request(1), "client"); + callbacks.trackSendFailure!( + request(1), + new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", { + timeout: 60_000, + }), + ); + expect(outstandingIds(client)).toEqual([1]); + }); + + it("only tracks client-originated requests, and ignores a rollback for an unknown id", () => { + const { client, callbacks } = makeClient(); + callbacks.trackRequest!(request(7), "server"); + expect(outstandingIds(client)).toEqual([]); + let changes = 0; + client.addEventListener("connectionDiagnosticsChange", () => { + changes += 1; + }); + callbacks.trackSendFailure!(request(7), new Error("nope")); + expect(changes).toBe(0); + }); +}); diff --git a/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts b/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts index 1d59b96a..4265186f 100644 --- a/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts +++ b/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts @@ -48,6 +48,18 @@ describe("MessageTrackingTransport.send", () => { await expect(failing.send(request)).rejects.toBe(boom); expect(callbacks.trackRequest).toHaveBeenCalledWith(request, "client"); expect(trackSendFailure).toHaveBeenCalledWith(request, boom); + // A caller's own abort is a cancellation, not a send failure: the + // request may well have reached the server, and stays unanswered. + trackSendFailure.mockClear(); + const controller = new AbortController(); + controller.abort(); + await expect( + failing.send( + { jsonrpc: "2.0", id: 2, method: "tools/call" }, + { requestSignal: controller.signal }, + ), + ).rejects.toBe(boom); + expect(trackSendFailure).not.toHaveBeenCalled(); // Only requests roll back: a failed notification has no pending entry. trackSendFailure.mockClear(); await expect( diff --git a/clients/web/src/test/integration/mcp/node/transport.test.ts b/clients/web/src/test/integration/mcp/node/transport.test.ts index c813fdf0..e5326b52 100644 --- a/clients/web/src/test/integration/mcp/node/transport.test.ts +++ b/clients/web/src/test/integration/mcp/node/transport.test.ts @@ -188,6 +188,35 @@ describe("Transport", () => { } }); + it("tracks fetches when only onFetchResponseBody is supplied (#2318)", async () => { + // Any one of the tracking callbacks earns the tracker — a caller that + // wants only bodies (or only stream updates) must not be handed the + // bare fetch and silently receive nothing. + const server = createTestServerHttp({ + serverInfo: createTestServerInfo(), + tools: [createEchoTool()], + serverType: "streamable-http", + }); + try { + await server.start(); + const bodies: string[] = []; + const result = createTransportNode( + { type: "streamable-http", url: server.url }, + { onFetchResponseBody: (_id, body) => bodies.push(body) }, + ); + const client = new Client( + { name: "test-client", version: "1.0.0" }, + { capabilities: {} }, + ); + await client.connect(result.transport); + await client.listTools(); + await client.close(); + expect(bodies.some((body) => body.includes("tools"))).toBe(true); + } finally { + await server.stop(); + } + }); + it("should call onFetchRequest callback for streamable-http transport", async () => { const server = createTestServerHttp({ serverInfo: createTestServerInfo(), diff --git a/clients/web/src/test/integration/mcp/remote/remoteClientTransport-unit.test.ts b/clients/web/src/test/integration/mcp/remote/remoteClientTransport-unit.test.ts index ef7ebf5f..e4656e4d 100644 --- a/clients/web/src/test/integration/mcp/remote/remoteClientTransport-unit.test.ts +++ b/clients/web/src/test/integration/mcp/remote/remoteClientTransport-unit.test.ts @@ -16,6 +16,7 @@ import type { RemoteTransportOptions } from "@inspector/core/mcp/remote/remoteCl import type { MCPServerConfig } from "@inspector/core/mcp/types.js"; import type { RemoteEvent } from "@inspector/core/mcp/remote/types.js"; import type { JSONRPCMessage } from "@modelcontextprotocol/client"; +import { SdkErrorCode } from "@modelcontextprotocol/client"; import { AuthChallengeError, AuthRecoveryRequiredError, @@ -615,6 +616,27 @@ describe("RemoteClientTransport (focused branch coverage)", () => { ).rejects.toThrow(/Transport is closed/); }); + it("rejects an expired SSE response wait with the SDK's own timeout shape (#2318)", async () => { + // The relay's budget, not the request's, can be the one that expires; + // shaped as the SDK's timeout so InspectorClient annotates it the same. + const t = makeTransport( + { + events: () => openEventsResponse(), + send: () => jsonResponse({ ok: true }), + }, + { sseResponseTimeoutMs: 50 }, + ); + await t.start(); + await expect( + t.send({ jsonrpc: "2.0", id: 1, method: "tools/list" }), + ).rejects.toMatchObject({ + code: SdkErrorCode.RequestTimeout, + message: "Request timed out", + data: { timeout: 50 }, + }); + await t.close(); + }); + it("does not wait for an SSE response on subscriptions/listen (long-lived stream)", async () => { // A `subscriptions/listen` never yields a JSON-RPC response — only an // acknowledged notification over the event channel — so send() must diff --git a/core/mcp/fetchTracking.ts b/core/mcp/fetchTracking.ts index 1928d1b7..fb3f117f 100644 --- a/core/mcp/fetchTracking.ts +++ b/core/mcp/fetchTracking.ts @@ -222,11 +222,13 @@ export function isLongLivedStreamResponse( method: string, contentType: string | null | undefined, ): boolean { - if (method !== "GET") return false; + if (method.toUpperCase() !== "GET") return false; if (!contentType) return false; + // Media types are case-insensitive (RFC 9110 §8.3.1); a server may send + // `Text/Event-Stream` and still be sending an event stream. + const type = contentType.toLowerCase(); return ( - contentType.includes("text/event-stream") || - contentType.includes("application/x-ndjson") + type.includes("text/event-stream") || type.includes("application/x-ndjson") ); } @@ -261,7 +263,9 @@ export type LongLivedStreamFraming = "sse" | "ndjson"; export function longLivedStreamFraming( contentType: string | null | undefined, ): LongLivedStreamFraming { - return contentType?.includes("application/x-ndjson") ? "ndjson" : "sse"; + return contentType?.toLowerCase().includes("application/x-ndjson") + ? "ndjson" + : "sse"; } /** @@ -314,8 +318,7 @@ function watchLongLivedStream( // stream with no state to show — the count only moves on an event. update(id, { eventCount }); - const consumeLine = (rawLine: string): void => { - const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + const consumeLine = (line: string): void => { if (framing === "ndjson") { if (line.trim() === "") return; eventCount += 1; @@ -335,18 +338,35 @@ function watchLongLivedStream( if (line === "data" || line.startsWith("data:")) blockHasData = true; }; + // Split off every complete line in `pending`. The SSE grammar (and the + // SDK's parser) accept LF, CRLF and a bare CR as a line ending, so all three + // end a line here; a CR that is the last byte so far is held back, since + // the LF of a CRLF pair may still be in flight in the next chunk. + const consumeCompleteLines = (): void => { + for (;;) { + const cr = pending.indexOf("\r"); + const lf = pending.indexOf("\n"); + const end = cr === -1 ? lf : lf === -1 ? cr : Math.min(cr, lf); + if (end === -1) return; + if (pending[end] === "\r") { + if (end === pending.length - 1) return; + const width = pending[end + 1] === "\n" ? 2 : 1; + consumeLine(pending.slice(0, end)); + pending = pending.slice(end + width); + } else { + consumeLine(pending.slice(0, end)); + pending = pending.slice(end + 1); + } + } + }; + const pump = async (): Promise => { try { for (;;) { const { value, done } = await reader.read(); if (done) break; pending += decoder.decode(value, { stream: true }); - let newline = pending.indexOf("\n"); - while (newline !== -1) { - consumeLine(pending.slice(0, newline)); - pending = pending.slice(newline + 1); - newline = pending.indexOf("\n"); - } + consumeCompleteLines(); } } catch { // An errored or aborted stream has still ended — fall through to report @@ -384,7 +404,12 @@ export function createFetchTracker( : input instanceof URL ? input.toString() : input.url; - const method = init?.method || "GET"; + // The method decides the long-lived-stream classification below, so it + // has to be the one the request actually carries: a `Request` input + // carries its own, and HTTP methods compare case-insensitively. + const method = ( + init?.method ?? (input instanceof Request ? input.method : "GET") + ).toUpperCase(); // Extract headers, redacting sensitive values BEFORE they reach any // downstream sink (logger, in-memory list, persisted session storage). diff --git a/core/mcp/inspectorClient.ts b/core/mcp/inspectorClient.ts index ca5c80ba..72bafec9 100644 --- a/core/mcp/inspectorClient.ts +++ b/core/mcp/inspectorClient.ts @@ -273,6 +273,7 @@ import { import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; import { annotateRequestTimeout, + isRequestTimeoutError, type ConnectionDiagnostics, type LastResponse, type NotificationStreamState, @@ -1131,11 +1132,16 @@ export class InspectorClient extends InspectorClientEventTarget { }; this.dispatchTypedEvent("message", entry); }, - trackSendFailure: (message: JSONRPCRequest) => { - // The frame never reached the wire, so it is not awaiting a response. - // Leaving it would report a request the server never saw as - // unanswered, in every timeout message and Connection Info row, until - // the session resets (#2318). + trackSendFailure: (message: JSONRPCRequest, error: unknown) => { + // A send that failed because the request timed out is not a send + // that failed: the browser's remote transport awaits the response + // inside `send`, so its relay timeout surfaces here — the request + // reached the server and went unanswered, which is the state to + // keep reporting. Every other failure means the frame never reached + // the wire, and leaving it would report a request the server never + // saw as unanswered, in every timeout message and Connection Info + // row, until the session resets (#2318). + if (isRequestTimeoutError(error)) return; if (this.outstandingRequests.delete(message.id)) { this.dispatchConnectionDiagnosticsChange(); } diff --git a/core/mcp/messageTrackingTransport.ts b/core/mcp/messageTrackingTransport.ts index b9dea9ca..4a94df37 100644 --- a/core/mcp/messageTrackingTransport.ts +++ b/core/mcp/messageTrackingTransport.ts @@ -29,7 +29,11 @@ export interface MessageTrackingCallbacks { * closed, the fetch failed before the frame reached the wire). Fires after * `trackRequest` already recorded it, so a consumer keeping "requests still * awaiting a response" can roll that entry back — a request that never went - * out is not unanswered (#2318). + * out is not unanswered (#2318). Not fired when the caller aborted the + * request through `requestSignal`: the modern transport aborts an in-flight + * request's own stream to cancel it (and the SDK does the same on a + * per-request timeout), and a request that was sent and then given up on + * is exactly the unanswered kind. */ trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void; } @@ -106,7 +110,9 @@ export class MessageTrackingTransport implements Transport { const request = message as JSONRPCRequest; this.callbacks.trackRequest?.(request, "client"); return this.baseTransport.send(message, options).catch((err) => { - this.callbacks.trackSendFailure?.(request, err); + if (!options?.requestSignal?.aborted) { + this.callbacks.trackSendFailure?.(request, err); + } throw err; }); } diff --git a/core/mcp/node/transport.ts b/core/mcp/node/transport.ts index e0dadcbf..c9a8ee54 100644 --- a/core/mcp/node/transport.ts +++ b/core/mcp/node/transport.ts @@ -55,6 +55,12 @@ export function createTransportNode( interceptAuthChallenges = false, onAuthChallengeObserved, } = options; + // Any of the three tracking callbacks earns a tracker: each is optional on + // the tracker itself, and a caller that wants only stream updates (or only + // bodies) must not be silently handed the bare fetch (#2318). + const wantsFetchTracking = Boolean( + onFetchRequest || onFetchResponseBody || onFetchStreamUpdate, + ); // `optionsFetchFn` is the caller's whole fetch stack and already sits on top of // a proxy-aware base when one is needed — the Node clients install @@ -115,7 +121,7 @@ export function createTransportNode( const sseFetch = configuredSseFetch ? withChallengeObserver(configuredSseFetch) : fetchWithOptionalAuthIntercept; - const trackedFetch = onFetchRequest + const trackedFetch = wantsFetchTracking ? createFetchTracker(sseFetch, { trackRequest: onFetchRequest, updateResponseBody: onFetchResponseBody, @@ -136,7 +142,7 @@ export function createTransportNode( ...(headers && { headers }), }; - const postFetch = onFetchRequest + const postFetch = wantsFetchTracking ? createFetchTracker(fetchWithOptionalAuthIntercept, { trackRequest: onFetchRequest, updateResponseBody: onFetchResponseBody, @@ -164,7 +170,7 @@ export function createTransportNode( ...(headers && { headers }), }; - const transportFetch = onFetchRequest + const transportFetch = wantsFetchTracking ? createFetchTracker(fetchWithOptionalAuthIntercept, { trackRequest: onFetchRequest, updateResponseBody: onFetchResponseBody, diff --git a/core/mcp/remote/remoteClientTransport.ts b/core/mcp/remote/remoteClientTransport.ts index 6a0ba873..a5c918b9 100644 --- a/core/mcp/remote/remoteClientTransport.ts +++ b/core/mcp/remote/remoteClientTransport.ts @@ -11,6 +11,7 @@ import type { JSONRPCMessage, MessageExtraInfo, } from "@modelcontextprotocol/client"; +import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; import type { InspectorServerSettings, StderrLogEntry } from "../types.js"; import type { FetchRequestEntryBase, FetchStreamState } from "../types.js"; import type { @@ -689,10 +690,17 @@ export class RemoteClientTransport implements Transport { return new Promise((resolve, reject) => { const timer = setTimeout(() => { this.sseResponseWaits.delete(requestId); + // The SDK's own per-request timeout shape, not a plain `Error`: this + // wait can expire before the SDK's timer does (its budget is the + // relay's, not the request's), and the rejection travels up through + // `Client.request`, where `InspectorClient` annotates a timeout of + // exactly this shape with the connection's state (#2318). A plain + // error would pass that decorator untouched and reach the user as a + // bare timeout on the web client alone. reject( - new Error( - `Timed out waiting for MCP response on SSE (${this.sseResponseTimeoutMs}ms)`, - ), + new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", { + timeout: this.sseResponseTimeoutMs, + }), ); }, this.sseResponseTimeoutMs); this.sseResponseWaits.set(requestId, {