diff --git a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx index 8315db77..b19fe362 100644 --- a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx +++ b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx @@ -17,6 +17,7 @@ import { RiErrorWarningLine } from "react-icons/ri"; import type { FetchRequestEntry } from "@inspector/core/mcp/types.js"; import { isLongLivedStreamEntry, + longLivedStreamLabel, uncapturedBodyNote, } from "../../../utils/uncapturedBodyNote"; import { ContentViewer } from "../../elements/ContentViewer/ContentViewer"; @@ -549,7 +550,9 @@ export function NetworkEntry({ {entry.duration != null && ( {formatDuration(entry.duration)} )} - {isLongLivedStreamEntry(entry) && SSE} + {isLongLivedStreamEntry(entry) && ( + {longLivedStreamLabel(entry)} + )} {statusLabel(entry)} diff --git a/clients/web/src/test/core/mcp/fetchTracking.test.ts b/clients/web/src/test/core/mcp/fetchTracking.test.ts index ae77fd26..61dc817e 100644 --- a/clients/web/src/test/core/mcp/fetchTracking.test.ts +++ b/clients/web/src/test/core/mcp/fetchTracking.test.ts @@ -792,6 +792,40 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]); }); + it("counts every non-blank line on an NDJSON stream", async () => { + const server = controlledStream(); + const updates: Array<{ id: string; stream: FetchStreamState }> = []; + const fetcher = createFetchTracker( + (async () => + new Response(server.stream, { + headers: { "content-type": "application/x-ndjson" }, + })) as typeof fetch, + { updateStream: (id, state) => updates.push({ id, stream: state }) }, + ); + await fetcher("https://example.com/mcp", { method: "GET" }); + server.push('{"jsonrpc":"2.0","method":"a"}\n\n'); + server.push('{"jsonrpc":"2.0","method":"b"}\r\n'); + // A partial line is not an event until its newline arrives. + server.push('{"jsonrpc":"2.0",'); + await flush(); + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]); + server.push('"method":"c"}\n'); + await flush(); + expect(updates.at(-1)!.stream.eventCount).toBe(3); + }); + + it("drops a partial SSE block at end-of-stream, as the SDK's parser does", async () => { + const server = controlledStream(); + const { updates } = await trackStream(server.stream); + // A data line with no dispatching blank line before the stream ends. + server.push("data: one\n\ndata: two\n"); + server.end(); + await flush(); + const last = updates.at(-1)!.stream; + expect(last.eventCount).toBe(1); + expect(last.closedAt).toBeInstanceOf(Date); + }); + 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-raw-wire.test.ts b/clients/web/src/test/core/mcp/inspectorClient-raw-wire.test.ts index 2d9af6b1..c6679c3f 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,4 +1,5 @@ import { describe, it, expect, vi } from "vitest"; +import { SdkErrorCode } from "@modelcontextprotocol/client"; import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js"; import { ModernGetTaskResultSchema } from "@inspector/core/mcp/modernTaskSchemas.js"; @@ -134,7 +135,15 @@ describe("InspectorClient raw-wire channel (#1631)", () => { {}, ModernGetTaskResultSchema, ); - const assertion = expect(promise).rejects.toThrow(/timed out/); + // The same annotated shape an SDK request's timeout carries (#2318): + // the SDK's code, the budget, the method, and the connection state. + const assertion = expect(promise).rejects.toMatchObject({ + code: SdkErrorCode.RequestTimeout, + message: expect.stringMatching( + /^Request timed out after 10ms \(tasks\/get\)\. No requests are unanswered\./, + ), + data: expect.objectContaining({ timeout: 10, method: "tasks/get" }), + }); await vi.advanceTimersByTimeAsync(20); await assertion; } finally { diff --git a/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts b/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts index 5f7204a2..1d59b96a 100644 --- a/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts +++ b/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts @@ -33,6 +33,30 @@ function makeTracked() { } describe("MessageTrackingTransport.send", () => { + it("reports a request the base transport failed to send, after tracking it (#2318)", async () => { + const { callbacks, base, tracked } = makeTracked(); + const trackSendFailure = vi.fn(); + const failing = new MessageTrackingTransport(base, { + ...callbacks, + trackSendFailure, + }); + const boom = new Error("connection closed"); + base.send = async () => { + throw boom; + }; + const request = { jsonrpc: "2.0", id: 1, method: "tools/list" } as const; + await expect(failing.send(request)).rejects.toBe(boom); + expect(callbacks.trackRequest).toHaveBeenCalledWith(request, "client"); + expect(trackSendFailure).toHaveBeenCalledWith(request, boom); + // Only requests roll back: a failed notification has no pending entry. + trackSendFailure.mockClear(); + await expect( + failing.send({ jsonrpc: "2.0", method: "notifications/initialized" }), + ).rejects.toBe(boom); + expect(trackSendFailure).not.toHaveBeenCalled(); + void tracked; + }); + it("tracks an outgoing request", async () => { const { callbacks, tracked } = makeTracked(); const request = { jsonrpc: "2.0", id: 1, method: "tools/list" } as const; diff --git a/clients/web/src/test/core/mcp/remote/sessionStorage.test.ts b/clients/web/src/test/core/mcp/remote/sessionStorage.test.ts index a8fa25bb..a0b5249f 100644 --- a/clients/web/src/test/core/mcp/remote/sessionStorage.test.ts +++ b/clients/web/src/test/core/mcp/remote/sessionStorage.test.ts @@ -110,6 +110,45 @@ describe("RemoteInspectorClientStorage", () => { ); }); + it("loadSession revives a stream's closedAt alongside the entry timestamp (#2318)", async () => { + const fetchFn = vi.fn( + async () => + new Response( + JSON.stringify({ + fetchRequests: [ + { + id: "open", + timestamp: "2026-01-01T00:00:00.000Z", + method: "GET", + url: "http://example.com/mcp", + requestHeaders: {}, + stream: { eventCount: 2 }, + }, + { + id: "closed", + timestamp: "2026-01-01T00:00:00.000Z", + method: "GET", + url: "http://example.com/mcp", + requestHeaders: {}, + stream: { eventCount: 3, closedAt: "2026-01-01T00:00:05.000Z" }, + }, + ], + createdAt: 1735689600000, + updatedAt: 1735689600001, + }), + { status: 200 }, + ), + ); + const storage = makeStorage(fetchFn); + const state = await storage.loadSession("sid"); + // An open stream's state passes through untouched. + expect(state?.fetchRequests[0]?.stream).toEqual({ eventCount: 2 }); + const closed = state?.fetchRequests[1]?.stream; + expect(closed?.eventCount).toBe(3); + expect(closed?.closedAt).toBeInstanceOf(Date); + expect(closed?.closedAt?.toISOString()).toBe("2026-01-01T00:00:05.000Z"); + }); + it("loadSession preserves Date timestamps already provided as Date instances", async () => { const fetchFn = vi.fn( async () => diff --git a/clients/web/src/test/core/react/useInspectorClient.test.tsx b/clients/web/src/test/core/react/useInspectorClient.test.tsx index 613d329b..107ba45e 100644 --- a/clients/web/src/test/core/react/useInspectorClient.test.tsx +++ b/clients/web/src/test/core/react/useInspectorClient.test.tsx @@ -90,9 +90,9 @@ describe("useInspectorClient", () => { const client = new FakeInspectorClient(); // `getConnectionDiagnostics` is optional on the protocol; a double that // predates it must still yield the one shape consumers expect. - ( - client as unknown as { getConnectionDiagnostics?: unknown } - ).getConnectionDiagnostics = undefined; + Object.defineProperty(client, "getConnectionDiagnostics", { + value: undefined, + }); const { result } = renderHook(() => useInspectorClient(client)); expect(result.current.connectionDiagnostics).toEqual({ capturedAt: 0, diff --git a/clients/web/src/test/integration/mcp/inspectorClient-timeout-diagnostics.test.ts b/clients/web/src/test/integration/mcp/inspectorClient-timeout-diagnostics.test.ts index 7e3b1172..fd4649b6 100644 --- a/clients/web/src/test/integration/mcp/inspectorClient-timeout-diagnostics.test.ts +++ b/clients/web/src/test/integration/mcp/inspectorClient-timeout-diagnostics.test.ts @@ -271,6 +271,21 @@ describe("InspectorClient request-timeout diagnostics (#2318)", () => { const stream = c.getConnectionDiagnostics().notificationStream!; expect(stream.url).toBe(server.url); expect(stream.closedAt).toBeUndefined(); + // Open from when the headers arrived, not from when the GET went out. + const getEntryAtOpen = log + .getFetchRequests() + .find((entry) => entry.method === "GET")!; + expect(stream.openedAt).toBe( + getEntryAtOpen.timestamp.getTime() + (getEntryAtOpen.duration ?? 0), + ); + // The snapshot is a copy: mutating it does not reach the client. + const snapshot = c.getConnectionDiagnostics(); + snapshot.outstandingRequests.push({ + id: "x", + method: "bogus", + sentAt: 0, + }); + expect(c.getConnectionDiagnostics().outstandingRequests).toEqual([]); // A server push on the notification stream is an event. server.pushEvent({ diff --git a/clients/web/src/utils/uncapturedBodyNote.test.ts b/clients/web/src/utils/uncapturedBodyNote.test.ts index d907286d..1bb88b0e 100644 --- a/clients/web/src/utils/uncapturedBodyNote.test.ts +++ b/clients/web/src/utils/uncapturedBodyNote.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import type { FetchRequestEntry } from "@inspector/core/mcp/types.js"; import { isLongLivedStreamEntry, + longLivedStreamLabel, uncapturedBodyNote, } from "./uncapturedBodyNote"; @@ -26,6 +27,27 @@ describe("isLongLivedStreamEntry", () => { isLongLivedStreamEntry({ ...streamEntry, responseHeaders: undefined }), ).toBe(false); }); + + it("reads the content type whatever casing the entry recorded it with", () => { + expect( + isLongLivedStreamEntry({ + ...streamEntry, + responseHeaders: { "Content-Type": "text/event-stream" }, + }), + ).toBe(true); + }); +}); + +describe("longLivedStreamLabel", () => { + it("names the framing the stream uses", () => { + expect(longLivedStreamLabel(streamEntry)).toBe("SSE"); + expect( + longLivedStreamLabel({ + ...streamEntry, + responseHeaders: { "Content-Type": "application/x-ndjson" }, + }), + ).toBe("NDJSON"); + }); }); describe("uncapturedBodyNote", () => { diff --git a/clients/web/src/utils/uncapturedBodyNote.ts b/clients/web/src/utils/uncapturedBodyNote.ts index 636c7f33..e8e886fb 100644 --- a/clients/web/src/utils/uncapturedBodyNote.ts +++ b/clients/web/src/utils/uncapturedBodyNote.ts @@ -1,17 +1,38 @@ // The Network entry's body placeholder for a response whose body was not // captured (#2318). Pure: a string from the entry alone. -import { isLongLivedStreamResponse } from "@inspector/core/mcp/fetchTracking.js"; +import { + findHeader, + isLongLivedStreamResponse, + longLivedStreamFraming, +} from "@inspector/core/mcp/fetchTracking.js"; import type { FetchRequestEntry } from "@inspector/core/mcp/types.js"; -/** Whether the entry is an unbounded server-to-client stream (never buffered). */ +/** + * Whether the entry is an unbounded server-to-client stream (never + * buffered). The header lookup is case-insensitive: a recorded entry keeps + * the wire casing (`Content-Type` on some hosts, and in a restored session), + * and a miss here would silently reclassify the stream as a bounded body. + */ export function isLongLivedStreamEntry(entry: FetchRequestEntry): boolean { return isLongLivedStreamResponse( entry.method, - entry.responseHeaders?.["content-type"], + findHeader(entry.responseHeaders, "content-type"), ); } +/** + * The badge label for a long-lived stream entry — `SSE` or `NDJSON`, by the + * same content-type split the tracker counts events with. + */ +export function longLivedStreamLabel(entry: FetchRequestEntry): string { + return longLivedStreamFraming( + findHeader(entry.responseHeaders, "content-type"), + ) === "ndjson" + ? "NDJSON" + : "SSE"; +} + /** * What stands in for a body that was not captured. A long-lived stream is * never buffered, so its placeholder reports what the tracker learned by diff --git a/core/mcp/fetchTracking.ts b/core/mcp/fetchTracking.ts index fa4b63bd..1928d1b7 100644 --- a/core/mcp/fetchTracking.ts +++ b/core/mcp/fetchTracking.ts @@ -251,15 +251,32 @@ export interface FetchTrackingCallbacks { } /** - * Count the SSE events on a long-lived stream without buffering it, and - * report when it ends. + * The framing a long-lived stream's events arrive in — the two content types + * `isLongLivedStreamResponse` accepts. SSE dispatches on a blank line; + * NDJSON dispatches on every line. + */ +export type LongLivedStreamFraming = "sse" | "ndjson"; + +/** The framing for a long-lived stream's content type. */ +export function longLivedStreamFraming( + contentType: string | null | undefined, +): LongLivedStreamFraming { + return contentType?.includes("application/x-ndjson") ? "ndjson" : "sse"; +} + +/** + * Count the events on a long-lived stream without buffering it, and report + * when it ends. * * Reads a `clone()` of the response through a reader and discards the bytes, * so the transport keeps consuming the original at its own pace (the same - * tee the bounded-body path relies on). Counting follows the SSE framing the - * transport's own parser applies: a blank line dispatches the block before - * it, and only a block carrying a `data` field is an event — keepalive - * comments (`: ping`) and bare `event:`/`id:` lines are not. + * tee the bounded-body path relies on). Counting follows the framing the + * transport's own parser applies. For SSE a blank line dispatches the block + * before it, and only a block carrying a `data` field is an event — + * keepalive comments (`: ping`) and bare `event:`/`id:` lines are not; a + * partial block at end-of-stream is dropped, exactly as the SSE spec and the + * SDK's `EventSourceParserStream` (which has no flush) drop it. For NDJSON + * every non-blank line is one event. * * Reports once as soon as watching starts (zero events, open), once per event, * and once more on every exit from the read loop — a clean end-of-stream, a @@ -273,6 +290,7 @@ export interface FetchTrackingCallbacks { function watchLongLivedStream( response: Response, id: string, + framing: LongLivedStreamFraming, callbacks: FetchTrackingCallbacks, ): void { const update = callbacks.updateStream; @@ -298,6 +316,12 @@ function watchLongLivedStream( const consumeLine = (rawLine: string): void => { const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + if (framing === "ndjson") { + if (line.trim() === "") return; + eventCount += 1; + update(id, { eventCount }); + return; + } if (line === "") { if (blockHasData) { eventCount += 1; @@ -331,6 +355,10 @@ function watchLongLivedStream( } update(id, { 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 is wrapped in a `try` that treats an error as a + // close — so the promise cannot reject. void pump(); } @@ -481,7 +509,12 @@ export function createFetchTracker( // but its event count and its eventual close are exactly the state a // stalled connection is diagnosed from (#2318). if (isLongLivedStream && response.body && !response.bodyUsed) { - watchLongLivedStream(response, id, callbacks); + watchLongLivedStream( + response, + id, + longLivedStreamFraming(response.headers.get("content-type")), + callbacks, + ); } // Kick off a fire-and-forget read of the cloned body. The clone is an diff --git a/core/mcp/inspectorClient.ts b/core/mcp/inspectorClient.ts index 95b46f33..ca5c80ba 100644 --- a/core/mcp/inspectorClient.ts +++ b/core/mcp/inspectorClient.ts @@ -270,6 +270,7 @@ import { findHeader, isLongLivedStreamResponse, } from "./fetchTracking.js"; +import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; import { annotateRequestTimeout, type ConnectionDiagnostics, @@ -1130,6 +1131,15 @@ 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). + if (this.outstandingRequests.delete(message.id)) { + this.dispatchConnectionDiagnosticsChange(); + } + }, trackResponse: ( message: JSONRPCResultResponse | JSONRPCErrorResponse, origin: MessageOrigin, @@ -2915,8 +2925,19 @@ export class InspectorClient extends InspectorClientEventTarget { const raw = await new Promise((resolve, reject) => { const timer = setTimeout(() => { this.pendingRawWireRequests.delete(id); + // The same error, with the same annotation, as an SDK request that + // times out: this path bypasses `Protocol.request`, so it builds the + // SDK's own timeout shape and runs it through the decorator's + // annotation by hand — a raw-wire caller sees one kind of timeout, + // not two (#2318). reject( - new Error(`Raw request "${method}" timed out after ${timeoutMs} ms`), + annotateRequestTimeout( + new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", { + timeout: timeoutMs, + }), + method, + this.getConnectionDiagnostics(), + ), ); }, timeoutMs); this.pendingRawWireRequests.set(id, { resolve, reject, timer }); @@ -6083,7 +6104,9 @@ export class InspectorClient extends InspectorClientEventTarget { this.notificationStream = { id: entry.id, url: entry.url, - openedAt: entry.timestamp.getTime(), + // The entry's `timestamp` is when the request went out; the stream is + // open from when the response headers arrived, `duration` later. + openedAt: entry.timestamp.getTime() + (entry.duration ?? 0), eventCount: 0, }; this.dispatchConnectionDiagnosticsChange(); @@ -6098,7 +6121,12 @@ export class InspectorClient extends InspectorClientEventTarget { getConnectionDiagnostics(): ConnectionDiagnostics { return { capturedAt: Date.now(), - outstandingRequests: [...this.outstandingRequests.values()], + // Copies, not the tracked records: the snapshot is handed to callers + // and carried on errors, and a mutation there must not reach the + // correlation state. + outstandingRequests: [...this.outstandingRequests.values()].map( + (request) => ({ ...request }), + ), ...(this.lastResponse && { lastResponse: { ...this.lastResponse } }), ...(this.notificationStream && { notificationStream: this.publicStreamState(this.notificationStream), diff --git a/core/mcp/messageTrackingTransport.ts b/core/mcp/messageTrackingTransport.ts index 3402e244..b9dea9ca 100644 --- a/core/mcp/messageTrackingTransport.ts +++ b/core/mcp/messageTrackingTransport.ts @@ -24,6 +24,14 @@ export interface MessageTrackingCallbacks { message: JSONRPCNotification, origin: MessageOrigin, ) => void; + /** + * An outgoing request the base transport failed to send (the connection + * 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). + */ + trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void; } /** @@ -95,7 +103,12 @@ export class MessageTrackingTransport implements Transport { "client", ); } else if ("method" in message) { - this.callbacks.trackRequest?.(message as JSONRPCRequest, "client"); + const request = message as JSONRPCRequest; + this.callbacks.trackRequest?.(request, "client"); + return this.baseTransport.send(message, options).catch((err) => { + this.callbacks.trackSendFailure?.(request, err); + throw err; + }); } } else if ("method" in message) { this.callbacks.trackNotification?.( diff --git a/core/mcp/remote/sessionStorage.ts b/core/mcp/remote/sessionStorage.ts index 9c24ee59..7d3764fb 100644 --- a/core/mcp/remote/sessionStorage.ts +++ b/core/mcp/remote/sessionStorage.ts @@ -119,7 +119,10 @@ export class RemoteInspectorClientStorage implements InspectorClientStorage { const data = (await res.json()) as InspectorClientSessionState; - // Deserialize state (convert ISO strings back to Date objects) + // Deserialize state (convert ISO strings back to Date objects). The + // nested `stream.closedAt` is a Date too, and `JSON.stringify` turned it + // into an ISO string on the way out; left as a string it would fail the + // `getTime()` every consumer of the type performs (#2318). return { ...data, fetchRequests: data.fetchRequests.map((req) => ({ @@ -130,6 +133,9 @@ export class RemoteInspectorClientStorage implements InspectorClientStorage { : req.timestamp instanceof Date ? req.timestamp : new Date(req.timestamp), + ...(req.stream?.closedAt !== undefined && { + stream: { ...req.stream, closedAt: new Date(req.stream.closedAt) }, + }), })), }; }