diff --git a/clients/web/src/test/core/mcp/fetchTracking.test.ts b/clients/web/src/test/core/mcp/fetchTracking.test.ts index fef4764c..0419b840 100644 --- a/clients/web/src/test/core/mcp/fetchTracking.test.ts +++ b/clients/web/src/test/core/mcp/fetchTracking.test.ts @@ -934,6 +934,28 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { expect(updates.at(-1)!.stream.closedAt).toBeInstanceOf(Date); }); + it("returns the response even when the listener throws on the open report", async () => { + // The open is reported synchronously inside the fetch wrapper; a throw + // there must not turn a successful response into a rejected fetch. + const server = controlledStream(); + const response = new Response(server.stream, { + headers: { "content-type": "text/event-stream" }, + }); + const fetcher = createFetchTracker((async () => response) as typeof fetch, { + updateStream: () => { + throw new Error("sink closed"); + }, + streamUpdateIntervalMs: 0, + }); + await expect( + fetcher("https://example.com/mcp", { method: "GET" }), + ).resolves.toBe(response); + // ...and a throw from a coalesced count does not escape the timer either. + server.push("data: a\n\n"); + server.end(); + await flush(); + }); + it("survives a stream-update listener that throws on the close", async () => { // The watcher's promise is discarded, so a throw from the final report // would otherwise be an unhandled rejection — which fails the whole run. diff --git a/clients/web/src/test/core/mcp/inspectorClient-raw-wire.test.ts b/clients/web/src/test/core/mcp/inspectorClient-raw-wire.test.ts index c6679c3f..638891b5 100644 --- a/clients/web/src/test/core/mcp/inspectorClient-raw-wire.test.ts +++ b/clients/web/src/test/core/mcp/inspectorClient-raw-wire.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi } from "vitest"; -import { SdkErrorCode } from "@modelcontextprotocol/client"; +import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js"; import { ModernGetTaskResultSchema } from "@inspector/core/mcp/modernTaskSchemas.js"; @@ -151,6 +151,46 @@ describe("InspectorClient raw-wire channel (#1631)", () => { } }); + it("annotates a request timeout the transport's send rejects with (#2318)", async () => { + // The browser's remote transport awaits the response inside `send`, so + // its relay wait can expire before the local timer; it arrives here as + // the SDK's timeout shape and gets the same annotation. + const client = makeClient(); + internals(client).transport = { + send: vi.fn().mockRejectedValue( + new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", { + timeout: 60_000, + }), + ), + }; + await expect( + internals(client).rawWireRequest( + "tasks/get", + {}, + ModernGetTaskResultSchema, + ), + ).rejects.toMatchObject({ + code: SdkErrorCode.RequestTimeout, + message: expect.stringMatching( + /^Request timed out after 1m00s \(tasks\/get\)\. /, + ), + data: expect.objectContaining({ timeout: 60_000, method: "tasks/get" }), + }); + }); + + it("leaves a non-timeout send failure untouched", async () => { + const client = makeClient(); + const boom = new Error("connection closed"); + internals(client).transport = { send: vi.fn().mockRejectedValue(boom) }; + await expect( + internals(client).rawWireRequest( + "tasks/get", + {}, + ModernGetTaskResultSchema, + ), + ).rejects.toBe(boom); + }); + it("rejects all pending requests on teardown", async () => { const client = makeClient(); internals(client).transport = { 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 e4656e4d..8583b19f 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 @@ -375,6 +375,63 @@ describe("RemoteClientTransport (focused branch coverage)", () => { }); }); + it("reports a still-open stream closed on close(), since the backend's own close cannot arrive (#2318)", async () => { + const onFetchStreamUpdate = vi.fn(); + const { response, push } = createPushableEventStream(); + const t = makeTransport( + { events: () => response }, + { onFetchStreamUpdate, onFetchRequest: vi.fn() }, + ); + await t.start(); + const entry = (id: string, method: string, contentType: string) => + ({ + id, + method, + url: "http://x/mcp", + timestamp: new Date().toISOString(), + requestHeaders: {}, + responseStatus: 200, + responseHeaders: { "Content-Type": contentType }, + }) as never; + push({ + type: "fetch_request", + data: entry("stream", "GET", "text/event-stream"), + }); + push({ + type: "fetch_request", + data: entry("post", "POST", "text/event-stream"), + }); + push({ + type: "fetch_stream_update", + data: { id: "stream", eventCount: 2 }, + }); + push({ + type: "fetch_request", + data: entry("earlier", "GET", "text/event-stream"), + }); + push({ + type: "fetch_stream_update", + data: { + id: "earlier", + eventCount: 1, + closedAt: new Date().toISOString(), + }, + }); + await tick(); + onFetchStreamUpdate.mockClear(); + await t.close(); + // Only the stream still open is reported closed, with its last count; + // the bounded POST and the already-closed stream are not. + expect(onFetchStreamUpdate).toHaveBeenCalledTimes(1); + expect(onFetchStreamUpdate).toHaveBeenCalledWith("stream", { + eventCount: 2, + closedAt: expect.any(Date), + }); + // A second close reports nothing more. + await t.close(); + expect(onFetchStreamUpdate).toHaveBeenCalledTimes(1); + }); + it("drops fetch_stream_update events when no handler is registered", async () => { const onFetchRequest = vi.fn(); const t = makeTransport( diff --git a/core/mcp/fetchTracking.ts b/core/mcp/fetchTracking.ts index 19b00d23..10eb1d00 100644 --- a/core/mcp/fetchTracking.ts +++ b/core/mcp/fetchTracking.ts @@ -310,8 +310,21 @@ function watchLongLivedStream( framing: LongLivedStreamFraming, callbacks: FetchTrackingCallbacks, ): void { - const report = callbacks.updateStream; - if (!report) return; + const listener = callbacks.updateStream; + if (!listener) return; + // Every report goes through here: the listener is consumer code (a relay + // sink that may already be closed, a state store), and it runs from three + // places none of which may throw outward — synchronously inside the fetch + // wrapper (the open), from a timer (a coalesced count), and off the + // discarded watcher promise (the close). A throw is the listener's + // problem; the stream's bookkeeping carries on. + const report = (state: FetchStreamState): void => { + try { + listener(id, state); + } catch { + // See above. + } + }; const intervalMs = callbacks.streamUpdateIntervalMs ?? DEFAULT_STREAM_UPDATE_INTERVAL_MS; let body: ReadableStream | null; @@ -333,19 +346,19 @@ function watchLongLivedStream( let reportTimer: ReturnType | null = null; const reportCount = (): void => { if (intervalMs <= 0) { - report(id, { eventCount }); + report({ eventCount }); return; } if (reportTimer !== null) return; reportTimer = setTimeout(() => { reportTimer = null; - report(id, { eventCount }); + report({ eventCount }); }, intervalMs); }; // Announce the stream as open before a byte arrives. A stream that never // delivers anything is the #2187 tell, and it would otherwise be the one // stream with no state to show — the count only moves on an event. - report(id, { eventCount }); + report({ eventCount }); const consumeLine = (line: string): void => { if (framing === "ndjson") { @@ -410,17 +423,12 @@ function watchLongLivedStream( clearTimeout(reportTimer); reportTimer = null; } - try { - report(id, { eventCount, closedAt: new Date() }); - } catch { - // A listener that throws must not become an unhandled rejection off a - // promise nothing awaits; the stream is closed either way. - } + report({ eventCount, closedAt: new Date() }); }; // Deliberately not awaited: the fetch wrapper has to return the response // now, and this watcher lives as long as the stream does. It owns its own - // failures — the read loop and the final report are each wrapped in a - // `try` — so the promise cannot reject. + // failures — the read loop is wrapped in a `try`, and every report goes + // through the non-throwing `report` above — so the promise cannot reject. void pump(); } diff --git a/core/mcp/inspectorClient.ts b/core/mcp/inspectorClient.ts index 43b05f68..72f3c192 100644 --- a/core/mcp/inspectorClient.ts +++ b/core/mcp/inspectorClient.ts @@ -2980,7 +2980,17 @@ export class InspectorClient extends InspectorClientEventTarget { clearTimeout(pending.timer); this.pendingRawWireRequests.delete(id); } - reject(err instanceof Error ? err : new Error(String(err))); + // The browser's remote transport awaits the response inside `send`, + // so its relay wait can expire here first, as the SDK's timeout + // shape; annotate it exactly as the local timer above does (a + // non-timeout error passes through untouched). + reject( + annotateRequestTimeout( + err instanceof Error ? err : new Error(String(err)), + method, + this.getConnectionDiagnostics(), + ), + ); }); }); return resultSchema.parse(raw); diff --git a/core/mcp/remote/remoteClientTransport.ts b/core/mcp/remote/remoteClientTransport.ts index 8200b32a..202310c4 100644 --- a/core/mcp/remote/remoteClientTransport.ts +++ b/core/mcp/remote/remoteClientTransport.ts @@ -12,6 +12,7 @@ import type { MessageExtraInfo, } from "@modelcontextprotocol/client"; import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; +import { findHeader, isLongLivedStreamResponse } from "../fetchTracking.js"; import type { InspectorServerSettings, StderrLogEntry } from "../types.js"; import type { FetchRequestEntryBase, FetchStreamState } from "../types.js"; import type { @@ -265,6 +266,14 @@ export class RemoteClientTransport implements Transport { private eventStreamConsumeTask: Promise | null = null; private restartingEventStream = false; private closed = false; + /** + * Long-lived streams the backend has reported open, by fetch entry id, with + * the last event count seen for each. `close()` reports them closed + * itself: it aborts the event channel before asking the backend to + * disconnect, so the backend watcher's own close report lands on an + * unconsumed queue and is lost with the session (#2318). + */ + private openStreams = new Map(); private readonly sseResponseWaits = new Map< string | number, SseResponseWait @@ -548,6 +557,14 @@ export class RemoteClientTransport implements Transport { this.options.onFetchRequest ) { const entry = parsed.data; + if ( + isLongLivedStreamResponse( + entry.method, + findHeader(entry.responseHeaders, "content-type"), + ) + ) { + this.openStreams.set(entry.id, 0); + } this.options.onFetchRequest({ ...entry, timestamp: @@ -568,6 +585,10 @@ export class RemoteClientTransport implements Transport { this.options.onFetchStreamUpdate ) { const { id, eventCount, closedAt } = parsed.data; + if (closedAt !== undefined) this.openStreams.delete(id); + else if (this.openStreams.has(id)) { + this.openStreams.set(id, eventCount); + } this.options.onFetchStreamUpdate(id, { eventCount, ...(closedAt !== undefined && { closedAt: new Date(closedAt) }), @@ -856,6 +877,13 @@ export class RemoteClientTransport implements Transport { this.closed = true; this.cancelAllSseWaits(new Error("Transport closed")); await this.stopEventStream(); + // The disconnect below ends every stream the backend still holds open, + // and its close reports cannot reach us any more (see `openStreams`). + const closedAt = new Date(); + for (const [id, eventCount] of this.openStreams) { + this.options.onFetchStreamUpdate?.(id, { eventCount, closedAt }); + } + this.openStreams.clear(); if (this._sessionId) { try {