diff --git a/clients/web/src/App.test.tsx b/clients/web/src/App.test.tsx index 64bb3cfc..cc86b827 100644 --- a/clients/web/src/App.test.tsx +++ b/clients/web/src/App.test.tsx @@ -714,6 +714,7 @@ const DEFAULT_USE_INSPECTOR_CLIENT: ReturnType = { instructions: undefined, excludedTools: [], malformedListItems: [], + connectionDiagnostics: { capturedAt: 0, outstandingRequests: [] }, appRendererClient: null, connect: vi.fn().mockResolvedValue(undefined), disconnect: vi.fn().mockResolvedValue(undefined), diff --git a/clients/web/src/App.tsx b/clients/web/src/App.tsx index bc62376e..62959775 100644 --- a/clients/web/src/App.tsx +++ b/clients/web/src/App.tsx @@ -401,6 +401,7 @@ function App() { discoverResult, excludedTools, malformedListItems, + connectionDiagnostics, lastError, } = useInspectorClient(inspectorClient); // What every settings write that landed on disk actually wrote, so a later @@ -2028,6 +2029,7 @@ function App() { transport={connectionInfoTransport} protocolEra={protocolEra} discoverResult={discoverResult} + diagnostics={connectionDiagnostics} oauth={connectionInfoOAuth} onClearOAuth={ connectionInfoCanClearOAuth ? handleClearConnectionOAuth : undefined diff --git a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.stories.tsx b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.stories.tsx index 50549076..8e63e5a3 100644 --- a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.stories.tsx +++ b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.stories.tsx @@ -46,6 +46,51 @@ export const FullCapabilities: Story = { }, }; +// The snapshot clock, read once when the stories load. The rows start from it +// and then tick against the wall clock, so the fixture has to be "now" — a +// fixed historical date would read as months ago a second after mounting. +const STORY_NOW = Date.now(); + +/** + * The Connection Activity section, in the #2187 shape: the notification + * stream has been open the whole session and delivered nothing, `tools/list` + * has been outstanding for a minute, and nothing has been answered since + * `initialize`. The same facts a request timeout's message reports, shown + * while the request is still in flight (#2318). + */ +export const WithConnectionActivity: Story = { + args: { + initializeResult: fullResult, + clientCapabilities: fullClientCaps, + transport: "streamable-http", + diagnostics: { + capturedAt: STORY_NOW, + outstandingRequests: [ + { id: 2, method: "tools/list", sentAt: STORY_NOW - 60_000 }, + { id: 3, method: "ping", sentAt: STORY_NOW - 12_000 }, + ], + lastResponse: { method: "initialize", receivedAt: STORY_NOW - 61_000 }, + notificationStream: { + url: "http://127.0.0.1:9779/mcp", + openedAt: STORY_NOW - 252_000, + eventCount: 0, + }, + }, + }, + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await expect(canvas.getByText("Connection Activity")).toBeInTheDocument(); + // Ticking: the rows may have advanced a second or two since the fixture + // was stamped, so match the shape rather than the exact second. + await expect( + canvas.getByText(/^tools\/list — sent 1m0\ds ago$/), + ).toBeInTheDocument(); + await expect( + canvas.getByText(/^GET \/mcp — open for 4m1\ds, 0 events delivered$/), + ).toBeInTheDocument(); + }, +}; + export const ModernEra: Story = { args: { initializeResult: { diff --git a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.test.tsx b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.test.tsx index ba4fb132..00d6b61d 100644 --- a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.test.tsx +++ b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.test.tsx @@ -5,6 +5,13 @@ import type { InitializeResult, } from "@modelcontextprotocol/client"; import { renderWithMantine, screen } from "../../../test/renderWithMantine"; +import type { ConnectionDiagnostics } from "@inspector/core/mcp/connectionDiagnostics.js"; +import { + NO_NOTIFICATION_STREAM_LABEL, + NO_OUTSTANDING_REQUESTS_LABEL, + NO_RESPONSE_YET_LABEL, + NO_STREAM_ON_STDIO_LABEL, +} from "../../../utils/connectionActivity"; import { CLEAR_OAUTH_STATE_AND_DISCONNECT_LABEL, ConnectionInfoContent, @@ -934,3 +941,81 @@ describe("ConnectionInfoContent", () => { expect(onClearOAuth).toHaveBeenCalledTimes(1); }); }); + +describe("ConnectionInfoContent connection activity (#2318)", () => { + const NOW = 1_000_000; + const seconds = (n: number) => n * 1000; + + it("omits the section when no diagnostics are supplied", () => { + renderWithMantine( + , + ); + expect(screen.queryByText("Connection Activity")).not.toBeInTheDocument(); + }); + + it("renders the idle state: nothing unanswered, nothing received, stream not opened", () => { + renderWithMantine( + , + ); + expect(screen.getByText("Connection Activity")).toBeInTheDocument(); + expect(screen.getByText(NO_OUTSTANDING_REQUESTS_LABEL)).toBeInTheDocument(); + expect(screen.getByText(NO_RESPONSE_YET_LABEL)).toBeInTheDocument(); + expect(screen.getByText(NO_NOTIFICATION_STREAM_LABEL)).toBeInTheDocument(); + }); + + it("lists each unanswered request on its own line, with the last response and the open stream", () => { + const diagnostics: ConnectionDiagnostics = { + capturedAt: NOW, + outstandingRequests: [ + { id: 2, method: "tools/list", sentAt: NOW - seconds(60) }, + { id: 3, method: "ping", sentAt: NOW - seconds(12) }, + ], + lastResponse: { method: "initialize", receivedAt: NOW - seconds(61) }, + notificationStream: { + url: "http://127.0.0.1:9779/mcp", + openedAt: NOW - seconds(252), + eventCount: 0, + }, + }; + renderWithMantine( + , + ); + const outstanding = screen.getByTestId("connection-activity-outstanding"); + expect(outstanding.children).toHaveLength(2); + expect(outstanding.children[0]?.textContent).toBe( + "tools/list — sent 1m00s ago", + ); + expect(outstanding.children[1]?.textContent).toBe("ping — sent 12s ago"); + expect(screen.getByText("initialize — 1m01s ago")).toBeInTheDocument(); + expect( + screen.getByText("GET /mcp — open for 4m12s, 0 events delivered"), + ).toBeInTheDocument(); + }); + + it("reports the stream row as not applicable on stdio", () => { + renderWithMantine( + , + ); + // Both the Session row and the Notification stream row read N/A (stdio). + expect(screen.getAllByText(NO_STREAM_ON_STDIO_LABEL)).toHaveLength(2); + }); +}); diff --git a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx index 1c04c951..25290f67 100644 --- a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx +++ b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx @@ -18,6 +18,14 @@ import type { ServerCapabilities, } from "@modelcontextprotocol/client"; import type { ServerType } from "@inspector/core/mcp/types.js"; +import type { ConnectionDiagnostics } from "@inspector/core/mcp/connectionDiagnostics.js"; +import { + formatLastResponse, + formatNotificationStream, + formatOutstandingRequests, + NO_OUTSTANDING_REQUESTS_LABEL, +} from "../../../utils/connectionActivity"; +import { useTickingClock } from "../../../hooks/useTickingClock"; import { TASKS_EXTENSION_KEY } from "@inspector/core/mcp/modernTaskSchemas.js"; import { getSkillsExtension } from "@inspector/core/mcp/skills.js"; import type { OAuthClientRegistrationKind } from "@inspector/core/auth/types.js"; @@ -75,6 +83,16 @@ export interface ConnectionInfoContentProps { * capabilities, and extensions learned up front. Undefined on legacy. (#1626) */ discoverResult?: DiscoverResult; + /** + * What the client is still waiting on, when it last heard back, and the + * state of the notification stream (#2318). Renders the Connection Activity + * section when present — the same facts a request timeout reports, shown + * here while the request is still in flight. Durations first read against + * the snapshot's own `capturedAt` and then tick once a second while the + * panel is open (`useTickingClock`), so an in-flight request's age keeps + * moving rather than freezing at the moment it went out. + */ + diagnostics?: ConnectionDiagnostics; oauth?: OAuthDetails; onClearOAuth?: () => void; } @@ -291,6 +309,7 @@ export function ConnectionInfoContent({ transport, protocolEra, discoverResult, + diagnostics, oauth, onClearOAuth, }: ConnectionInfoContentProps) { @@ -317,6 +336,10 @@ export function ConnectionInfoContent({ ? serverInfo.version?.trim() || "—" : SERVER_INFO_NOT_REPORTED_LABEL; + // The activity rows' clock: the snapshot's own on first paint (pure), then + // the wall clock once a second so "sent 5s ago" keeps counting. + const now = useTickingClock(diagnostics?.capturedAt ?? 0); + const serverCaps = getServerCapabilityEntries(capabilities, protocolEra); const clientCaps = getCapabilityEntries( clientCapabilities, @@ -348,6 +371,42 @@ export function ConnectionInfoContent({ + {diagnostics && ( + + Connection Activity + + Unanswered requests + {/* A `Stack` so several in-flight requests read as a list, the + way the extension sections do; one line per request. */} + + {diagnostics.outstandingRequests.length === 0 ? ( + {NO_OUTSTANDING_REQUESTS_LABEL} + ) : ( + formatOutstandingRequests(diagnostics, now).map( + (line, index) => ( + + {line} + + ), + ) + )} + + + Last response + {formatLastResponse(diagnostics, now)} + + Notification stream + + {formatNotificationStream( + diagnostics.notificationStream, + transport, + now, + )} + + + + )} + {discoverResult && ( Discovery diff --git a/clients/web/src/components/groups/ConnectionInfoModal/ConnectionInfoModal.test.tsx b/clients/web/src/components/groups/ConnectionInfoModal/ConnectionInfoModal.test.tsx index c73acff5..041745f3 100644 --- a/clients/web/src/components/groups/ConnectionInfoModal/ConnectionInfoModal.test.tsx +++ b/clients/web/src/components/groups/ConnectionInfoModal/ConnectionInfoModal.test.tsx @@ -56,6 +56,27 @@ describe("ConnectionInfoModal", () => { expect(screen.getByText("Be excellent to each other.")).toBeInTheDocument(); }); + it("forwards connection diagnostics to ConnectionInfoContent (#2318)", () => { + renderWithMantine( + , + ); + expect(screen.getByText("Connection Activity")).toBeInTheDocument(); + expect(screen.getByText("tools/list — sent 1m00s ago")).toBeInTheDocument(); + }); + it("forwards oauth details to ConnectionInfoContent", () => { renderWithMantine( void; } @@ -51,6 +54,7 @@ export function ConnectionInfoModal({ transport, protocolEra, discoverResult, + diagnostics, oauth, onClearOAuth, }: ConnectionInfoModalProps) { @@ -79,6 +83,7 @@ export function ConnectionInfoModal({ transport={transport} protocolEra={protocolEra} discoverResult={discoverResult} + diagnostics={diagnostics} oauth={oauth} onClearOAuth={onClearOAuth} /> diff --git a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.stories.tsx b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.stories.tsx index 21349a41..725702d2 100644 --- a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.stories.tsx +++ b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.stories.tsx @@ -86,6 +86,31 @@ const streamingEntry: FetchRequestEntry = { category: "transport", }; +/** + * The standalone `GET /mcp` notification stream, still open, as the tracker + * watches it: the body is never captured, so the placeholder reports what the + * stream has delivered instead (#2318). + */ +const notificationStreamEntry: FetchRequestEntry = { + id: "n-notify", + timestamp: new Date("2026-03-17T10:30:01Z"), + method: "GET", + url: "http://localhost:3000/mcp", + requestHeaders: { + accept: "text/event-stream", + "mcp-session-id": "0a0b0a5-fd27-4c95-a805-c0fba67e00fb", + }, + responseStatus: 200, + responseStatusText: "OK", + responseHeaders: { + "cache-control": "no-cache", + "content-type": "text/event-stream", + }, + duration: 12, + stream: { eventCount: 0 }, + category: "transport", +}; + const transportError: FetchRequestEntry = { id: "n-4", timestamp: new Date("2026-03-17T10:30:15Z"), @@ -166,6 +191,18 @@ export const StreamingResponse: Story = { args: { entry: streamingEntry, isListExpanded: true }, }; +export const NotificationStreamOpen: Story = { + args: { entry: notificationStreamEntry, isListExpanded: true }, + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await expect( + canvas.getByText( + "Long-lived stream — 0 events delivered, still open; body not captured", + ), + ).toBeInTheDocument(); + }, +}; + export const FetchError: Story = { args: { entry: transportError, isListExpanded: true }, }; diff --git a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.test.tsx b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.test.tsx index e42d841c..e5fa1e70 100644 --- a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.test.tsx +++ b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.test.tsx @@ -150,6 +150,24 @@ describe("NetworkEntry", () => { ).toBeInTheDocument(); }); + it("shows the watched stream's event count and open state in place of the body (#2318)", async () => { + const user = userEvent.setup(); + const open: FetchRequestEntry = { + ...baseEntry, + method: "GET", + responseHeaders: { "content-type": "text/event-stream" }, + responseBody: undefined, + stream: { eventCount: 3 }, + }; + renderWithMantine(); + await user.click(screen.getByRole("button", { name: "Expand" })); + expect( + screen.getByText( + "Long-lived stream — 3 events delivered, still open; body not captured", + ), + ).toBeInTheDocument(); + }); + it("shows an SSE badge on long-lived GET event-stream entries", () => { const sse: FetchRequestEntry = { ...baseEntry, diff --git a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx index 8aecb0fc..bec6b08c 100644 --- a/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx +++ b/clients/web/src/components/groups/NetworkEntry/NetworkEntry.tsx @@ -15,7 +15,10 @@ import { } from "@mantine/core"; import { RiErrorWarningLine } from "react-icons/ri"; import type { FetchRequestEntry } from "@inspector/core/mcp/types.js"; -import { isLongLivedStreamResponse } from "@inspector/core/mcp/fetchTracking.js"; +import { + isLongLivedStreamEntry, + uncapturedBodyNote, +} from "../../../utils/uncapturedBodyNote"; import { ContentViewer } from "../../elements/ContentViewer/ContentViewer"; import { getMimeKind } from "../../elements/ContentViewer/contentViewerUtils"; import { CopyButton } from "../../elements/CopyButton/CopyButton"; @@ -158,13 +161,6 @@ function statusLabel(entry: FetchRequestEntry): string { : `${entry.responseStatus}`; } -function isLongLivedStream(entry: FetchRequestEntry): boolean { - return isLongLivedStreamResponse( - entry.method, - entry.responseHeaders?.["content-type"], - ); -} - // Header-table cell text. A modern MCP-mirrored header name gets a violet accent // so the spec headers (Mcp-Method / Mcp-Name / Mcp-Param-* / MCP-Protocol-Version) // stand out from ordinary ones; a value that disagrees with the request body is @@ -542,7 +538,7 @@ export function NetworkEntry({ {entry.duration != null && ( {formatDuration(entry.duration)} )} - {isLongLivedStream(entry) && SSE} + {isLongLivedStreamEntry(entry) && SSE} {statusLabel(entry)} @@ -647,11 +643,7 @@ export function NetworkEntry({ label="Response body" /> ) : ( - - {isLongLivedStream(entry) - ? "Long-lived stream — body not captured" - : "(empty)"} - + {uncapturedBodyNote(entry)} )} )} diff --git a/clients/web/src/hooks/useTickingClock.test.ts b/clients/web/src/hooks/useTickingClock.test.ts new file mode 100644 index 00000000..9242e1cb --- /dev/null +++ b/clients/web/src/hooks/useTickingClock.test.ts @@ -0,0 +1,58 @@ +import { describe, it, expect, vi, afterEach } from "vitest"; +import { act, renderHook } from "@testing-library/react"; +import { useTickingClock } from "./useTickingClock"; + +describe("useTickingClock", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it("reads the seed on the first render, before any tick", () => { + vi.useFakeTimers(); + vi.setSystemTime(5_000_000); + const { result } = renderHook(() => useTickingClock(1_000_000)); + expect(result.current).toBe(1_000_000); + }); + + it("advances to the wall clock on each interval", () => { + vi.useFakeTimers(); + vi.setSystemTime(5_000_000); + const { result } = renderHook(() => useTickingClock(1_000_000, 1000)); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(result.current).toBe(5_001_000); + act(() => { + vi.setSystemTime(5_010_000); + // Advancing the fake timers also advances the fake clock. + vi.advanceTimersByTime(1000); + }); + expect(result.current).toBe(5_011_000); + }); + + it("never reads behind a seed newer than the last tick", () => { + vi.useFakeTimers(); + vi.setSystemTime(5_000_000); + const { result, rerender } = renderHook( + ({ seed }) => useTickingClock(seed, 1000), + { initialProps: { seed: 1_000_000 } }, + ); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(result.current).toBe(5_001_000); + // A fresher snapshot arrives before the next tick: it is what the caller + // renders from, so it must not be measured against an older clock. + rerender({ seed: 5_002_500 }); + expect(result.current).toBe(5_002_500); + }); + + it("stops ticking on unmount", () => { + vi.useFakeTimers(); + const clearSpy = vi.spyOn(globalThis, "clearInterval"); + const { unmount } = renderHook(() => useTickingClock(0, 250)); + unmount(); + expect(clearSpy).toHaveBeenCalled(); + clearSpy.mockRestore(); + }); +}); diff --git a/clients/web/src/hooks/useTickingClock.ts b/clients/web/src/hooks/useTickingClock.ts new file mode 100644 index 00000000..d82ac3f7 --- /dev/null +++ b/clients/web/src/hooks/useTickingClock.ts @@ -0,0 +1,27 @@ +import { useEffect, useState } from "react"; + +/** + * A clock that starts at `seed` and advances to the wall clock once per + * `intervalMs` while the component is mounted (#2318). + * + * For rendering "N seconds ago" from a snapshot stamped with its own + * `capturedAt`: the first render reads the snapshot's clock (pure — no + * `Date.now()` in render, which `react-hooks/purity` reports), and the + * interval then keeps the displayed durations moving while the panel stays + * open. The seed also floors the result, so a snapshot newer than the last + * tick is never rendered against a clock that predates it — a duration + * cannot go negative, and a freshly captured snapshot reads as fresh until the + * next tick catches up. + * + * A timer is exactly the "synchronize with an external system" case an effect + * exists for; the state is set only from the interval callback, never in the + * effect body, so the first paint is not followed by a corrective re-render. + */ +export function useTickingClock(seed: number, intervalMs = 1000): number { + const [now, setNow] = useState(seed); + useEffect(() => { + const id = setInterval(() => setNow(Date.now()), intervalMs); + return () => clearInterval(id); + }, [intervalMs]); + return Math.max(now, seed); +} diff --git a/clients/web/src/test/core/mcp/connectionDiagnostics.test.ts b/clients/web/src/test/core/mcp/connectionDiagnostics.test.ts new file mode 100644 index 00000000..d86a111c --- /dev/null +++ b/clients/web/src/test/core/mcp/connectionDiagnostics.test.ts @@ -0,0 +1,279 @@ +import { describe, it, expect } from "vitest"; +import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; +import { + annotateRequestTimeout, + describeConnectionDiagnostics, + describeStreamEndpoint, + formatDuration, + isRequestTimeoutError, + MAX_LISTED_OUTSTANDING_REQUESTS, + type AnnotatedRequestTimeoutData, + type ConnectionDiagnostics, +} from "@inspector/core/mcp/connectionDiagnostics.js"; + +// A fixed clock so every duration in these assertions is exact. +const NOW = 1_000_000; +const seconds = (n: number) => n * 1000; + +/** The SDK's own per-request timeout, exactly as `Protocol.request` throws it. */ +function sdkTimeout(timeout = 60_000): SdkError { + return new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", { + timeout, + }); +} + +describe("formatDuration", () => { + it("renders sub-second values in milliseconds", () => { + expect(formatDuration(0)).toBe("0ms"); + expect(formatDuration(850)).toBe("850ms"); + expect(formatDuration(999.6)).toBe("1000ms"); + }); + + it("renders seconds, minutes and hours at increasing granularity", () => { + expect(formatDuration(seconds(1))).toBe("1s"); + expect(formatDuration(seconds(59))).toBe("59s"); + expect(formatDuration(seconds(60))).toBe("1m00s"); + expect(formatDuration(seconds(252))).toBe("4m12s"); + expect(formatDuration(seconds(3600))).toBe("1h00m"); + expect(formatDuration(seconds(3600 + 120 + 5))).toBe("1h02m"); + }); + + it("clamps a negative value to zero rather than rendering nonsense", () => { + // A clock skew between two timestamps must not print `-3s`. + expect(formatDuration(-3000)).toBe("0ms"); + }); +}); + +describe("describeStreamEndpoint", () => { + it("reduces a stream URL to its method and path", () => { + expect(describeStreamEndpoint("https://host.example:8443/mcp?x=1")).toBe( + "GET /mcp", + ); + }); + + it("falls back to the raw string for an unparseable URL", () => { + expect(describeStreamEndpoint("not a url")).toBe("GET not a url"); + }); +}); + +describe("describeConnectionDiagnostics", () => { + it("says so when nothing is outstanding and nothing has been received", () => { + expect( + describeConnectionDiagnostics({ + capturedAt: NOW, + outstandingRequests: [], + }), + ).toBe( + "No requests are unanswered. No response has been received this session.", + ); + }); + + it("names each unanswered request with how long ago it went out", () => { + const diagnostics: ConnectionDiagnostics = { + capturedAt: NOW, + outstandingRequests: [ + { id: 2, method: "tools/list", sentAt: NOW - seconds(60) }, + { id: 3, method: "ping", sentAt: NOW - seconds(12) }, + ], + lastResponse: { method: "initialize", receivedAt: NOW - seconds(61) }, + }; + expect(describeConnectionDiagnostics(diagnostics)).toBe( + "2 requests are unanswered: tools/list (sent 1m00s ago), ping (sent 12s ago). " + + "Last response received 1m01s ago (initialize).", + ); + }); + + it("uses the singular for exactly one unanswered request", () => { + const diagnostics: ConnectionDiagnostics = { + capturedAt: NOW, + outstandingRequests: [ + { id: 2, method: "tools/list", sentAt: NOW - seconds(60) }, + ], + }; + expect(describeConnectionDiagnostics(diagnostics)).toContain( + "1 request is unanswered: tools/list (sent 1m00s ago).", + ); + }); + + it("collapses the tail of a long outstanding list into a count", () => { + const outstandingRequests = Array.from({ length: 8 }, (_, i) => ({ + id: i, + method: `m${i}`, + sentAt: NOW - seconds(i + 1), + })); + const text = describeConnectionDiagnostics({ + capturedAt: NOW, + outstandingRequests, + }); + expect(text).toContain("8 requests are unanswered: "); + for (let i = 0; i < MAX_LISTED_OUTSTANDING_REQUESTS; i++) { + expect(text).toContain(`m${i} (sent`); + } + expect(text).not.toContain("m5 (sent"); + expect(text).toContain( + `, and ${8 - MAX_LISTED_OUTSTANDING_REQUESTS} more.`, + ); + }); + + it("describes an open notification stream by its endpoint, age and event count", () => { + const diagnostics: ConnectionDiagnostics = { + capturedAt: NOW, + outstandingRequests: [], + notificationStream: { + url: "http://127.0.0.1:9779/mcp", + openedAt: NOW - seconds(252), + eventCount: 0, + }, + }; + expect(describeConnectionDiagnostics(diagnostics)).toContain( + "Notification stream (GET /mcp) open for 4m12s, 0 events delivered.", + ); + }); + + it("describes a closed notification stream by when it closed and how long it lasted", () => { + const diagnostics: ConnectionDiagnostics = { + capturedAt: NOW, + outstandingRequests: [], + notificationStream: { + url: "http://127.0.0.1:9779/mcp", + openedAt: NOW - seconds(185), + eventCount: 1, + closedAt: NOW - seconds(5), + }, + }; + expect(describeConnectionDiagnostics(diagnostics)).toContain( + "Notification stream (GET /mcp) closed 5s ago after 3m00s, 1 event delivered.", + ); + }); + + it("measures every duration against the snapshot's own clock, not the wall clock", () => { + // A snapshot taken long ago still reads the way it read when it was + // taken — that is what makes a timeout message and a later render of the + // same snapshot agree. + const diagnostics: ConnectionDiagnostics = { + capturedAt: NOW, + outstandingRequests: [ + { id: 1, method: "tools/list", sentAt: NOW - seconds(5) }, + ], + }; + expect(describeConnectionDiagnostics(diagnostics)).toBe( + "1 request is unanswered: tools/list (sent 5s ago). " + + "No response has been received this session.", + ); + }); +}); + +describe("isRequestTimeoutError", () => { + it("recognises the SDK's per-request timeout", () => { + expect(isRequestTimeoutError(sdkTimeout())).toBe(true); + }); + + it("rejects the other RequestTimeout-coded rejections, which are not timeouts", () => { + // An abort through `options.signal` with a non-SdkError reason: the SDK + // wraps it in the same code with no `{ timeout }` data. + expect( + isRequestTimeoutError( + new SdkError(SdkErrorCode.RequestTimeout, "AbortError: cancelled"), + ), + ).toBe(false); + // The max-total-timeout variant carries different data. + expect( + isRequestTimeoutError( + new SdkError( + SdkErrorCode.RequestTimeout, + "Maximum total timeout exceeded", + { maxTotalTimeout: 1000, totalElapsed: 1200 }, + ), + ), + ).toBe(false); + // A non-numeric `timeout` is not the SDK's shape either. + expect( + isRequestTimeoutError( + new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", { + timeout: "60s", + }), + ), + ).toBe(false); + }); + + it("rejects other SdkErrors, plain errors and non-errors", () => { + expect( + isRequestTimeoutError( + new SdkError(SdkErrorCode.InvalidResult, "bad", { timeout: 1 }), + ), + ).toBe(false); + expect(isRequestTimeoutError(new Error("Request timed out"))).toBe(false); + expect(isRequestTimeoutError("Request timed out")).toBe(false); + expect(isRequestTimeoutError(null)).toBe(false); + }); +}); + +describe("annotateRequestTimeout", () => { + const diagnostics: ConnectionDiagnostics = { + capturedAt: NOW, + outstandingRequests: [ + { id: 2, method: "tools/list", sentAt: NOW - seconds(60) }, + ], + lastResponse: { method: "initialize", receivedAt: NOW - seconds(61) }, + notificationStream: { + url: "http://127.0.0.1:9779/mcp", + openedAt: NOW - seconds(62), + eventCount: 0, + }, + }; + + it("rewrites the message to name the request, its budget and the connection state", () => { + const annotated = annotateRequestTimeout( + sdkTimeout(60_000), + "tools/list", + diagnostics, + ); + expect(SdkError.isInstance(annotated)).toBe(true); + expect((annotated as SdkError).message).toBe( + "Request timed out after 1m00s (tools/list). " + + "1 request is unanswered: tools/list (sent 1m00s ago). " + + "Last response received 1m01s ago (initialize). " + + "Notification stream (GET /mcp) open for 1m02s, 0 events delivered.", + ); + }); + + it("keeps the code, carries the method and snapshot in data, and links the original as cause", () => { + const original = sdkTimeout(300); + const annotated = annotateRequestTimeout( + original, + "ping", + diagnostics, + ) as SdkError; + expect(annotated).not.toBe(original); + expect(annotated.code).toBe(SdkErrorCode.RequestTimeout); + expect(annotated.cause).toBe(original); + const data = annotated.data as AnnotatedRequestTimeoutData; + expect(data.timeout).toBe(300); + expect(data.method).toBe("ping"); + expect(data.diagnostics).toBe(diagnostics); + // The rebuilt error still satisfies the predicate, so a consumer that + // checks before and after this runs sees the same answer. + expect(isRequestTimeoutError(annotated)).toBe(true); + }); + + it("returns anything that is not the per-request timeout untouched", () => { + const plain = new Error("boom"); + expect(annotateRequestTimeout(plain, "ping", diagnostics)).toBe(plain); + const abort = new SdkError(SdkErrorCode.RequestTimeout, "cancelled"); + expect(annotateRequestTimeout(abort, "ping", diagnostics)).toBe(abort); + expect(annotateRequestTimeout(undefined, "ping", diagnostics)).toBe( + undefined, + ); + }); + + it("reads cleanly when nothing else is known about the connection", () => { + const annotated = annotateRequestTimeout(sdkTimeout(1000), "ping", { + capturedAt: NOW, + outstandingRequests: [], + }) as SdkError; + expect(annotated.message).toBe( + "Request timed out after 1s (ping). No requests are unanswered. " + + "No response has been received this session.", + ); + }); +}); diff --git a/clients/web/src/test/core/mcp/fetchTracking.test.ts b/clients/web/src/test/core/mcp/fetchTracking.test.ts index 76abcb0f..ae77fd26 100644 --- a/clients/web/src/test/core/mcp/fetchTracking.test.ts +++ b/clients/web/src/test/core/mcp/fetchTracking.test.ts @@ -1,13 +1,17 @@ import { describe, it, expect, vi } from "vitest"; import { createFetchTracker, + findHeader, redactSensitiveHeaders, redactBody, redactUrlQuery, REDACTED_HEADER_VALUE, REDACTED_VALUE, } from "@inspector/core/mcp/fetchTracking.js"; -import type { FetchRequestEntryBase } from "@inspector/core/mcp/types.js"; +import type { + FetchRequestEntryBase, + FetchStreamState, +} from "@inspector/core/mcp/types.js"; // The tracker fires `trackRequest` synchronously with an entry whose // responseBody is always undefined, then reads the body in the background @@ -696,3 +700,196 @@ describe("redactSensitiveHeaders", () => { expect(input.authorization).toBe("Bearer x"); }); }); + +describe("findHeader", () => { + it("looks a header up case-insensitively and tolerates a missing record", () => { + expect(findHeader({ "Content-Type": "text/plain" }, "content-type")).toBe( + "text/plain", + ); + expect(findHeader({ "content-type": "text/plain" }, "CONTENT-TYPE")).toBe( + "text/plain", + ); + expect(findHeader({ accept: "*/*" }, "content-type")).toBeUndefined(); + expect(findHeader(undefined, "content-type")).toBeUndefined(); + }); +}); + +describe("createFetchTracker long-lived stream watching (#2318)", () => { + const encoder = new TextEncoder(); + + /** + * A server-controlled SSE body: `push` writes a chunk, `end` closes it, and + * `fail` errors it — the three ways a real stream advances. + */ + function controlledStream() { + let controller!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start(c) { + controller = c; + }, + }); + return { + stream, + push: (text: string) => controller.enqueue(encoder.encode(text)), + end: () => controller.close(), + fail: (err: Error) => controller.error(err), + }; + } + + /** Track a GET whose response is the given stream, collecting stream updates. */ + async function trackStream(stream: ReadableStream) { + const updates: Array<{ id: string; stream: FetchStreamState }> = []; + const tracked: FetchRequestEntryBase[] = []; + const fetcher = createFetchTracker( + (async () => + new Response(stream, { + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch, + { + trackRequest: (entry) => tracked.push(entry), + updateStream: (id, state) => updates.push({ id, stream: state }), + }, + ); + const response = await fetcher("https://example.com/mcp", { + method: "GET", + }); + return { response, updates, id: tracked[0]!.id }; + } + + it("announces the stream as open with zero events as soon as it is watched", async () => { + const server = controlledStream(); + const { updates, id } = await trackStream(server.stream); + // Before a byte arrives — this is the state the #2187 stall shows. + expect(updates).toEqual([{ id, stream: { eventCount: 0 } }]); + }); + + it("counts each data-carrying SSE event as it is delivered", async () => { + const server = controlledStream(); + const { updates, id } = await trackStream(server.stream); + server.push('event: message\ndata: {"jsonrpc":"2.0","method":"a"}\n\n'); + await flush(); + expect(updates.at(-1)).toEqual({ id, stream: { eventCount: 1 } }); + // An event split across chunks is still one event. + server.push('data: {"jsonrpc"'); + server.push(':"2.0","method":"b"}\n'); + await flush(); + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1]); + server.push("\n"); + await flush(); + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]); + }); + + it("ignores keepalive comments and blocks with no data field, and accepts a bare `data` line and CRLF framing", async () => { + const server = controlledStream(); + const { updates } = await trackStream(server.stream); + // A comment-only block (the keepalive shape) and an id-only block are + // not events; a bare `data` (no colon) is; so is a CRLF-delimited block. + server.push(": keepalive\n\n"); + server.push("id: 7\n\n"); + server.push("data\n\n"); + server.push("data: x\r\n\r\n"); + await flush(); + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]); + }); + + it("reports the close when the server ends the stream", async () => { + const server = controlledStream(); + const { updates, id } = await trackStream(server.stream); + server.push("data: one\n\n"); + server.end(); + await flush(); + expect(updates).toHaveLength(3); + expect(updates[2]!.id).toBe(id); + expect(updates[2]!.stream.eventCount).toBe(1); + expect(updates[2]!.stream.closedAt).toBeInstanceOf(Date); + // Only the close carries `closedAt`. + expect(updates[0]!.stream.closedAt).toBeUndefined(); + expect(updates[1]!.stream.closedAt).toBeUndefined(); + }); + + it("reports the close when the stream errors, with the count so far", async () => { + const server = controlledStream(); + const { updates } = await trackStream(server.stream); + server.push("data: one\n\n"); + server.fail(new Error("connection reset")); + await flush(); + expect(updates).toHaveLength(3); + expect(updates[2]!.stream).toMatchObject({ eventCount: 1 }); + expect(updates[2]!.stream.closedAt).toBeInstanceOf(Date); + }); + + it("leaves the transport's own copy of the stream intact", async () => { + const server = controlledStream(); + const { response, updates } = await trackStream(server.stream); + server.push("data: hello\n\n"); + server.end(); + // The transport reads the original; the watcher read a tee'd clone. + expect(await response.text()).toBe("data: hello\n\n"); + await flush(); + expect(updates.at(-1)!.stream).toMatchObject({ eventCount: 1 }); + }); + + it("does not clone the response when nobody subscribed to stream updates", async () => { + const response = new Response(controlledStream().stream, { + headers: { "content-type": "text/event-stream" }, + }); + const clone = vi.spyOn(response, "clone"); + const fetcher = createFetchTracker((async () => response) as typeof fetch, { + trackRequest: () => {}, + }); + await fetcher("https://example.com/mcp", { method: "GET" }); + expect(clone).not.toHaveBeenCalled(); + }); + + it("does not watch a bounded response", async () => { + const updates: unknown[] = []; + const fetcher = createFetchTracker( + (async () => + new Response("data: x\n\n", { + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch, + { updateStream: (id, state) => updates.push({ id, state }) }, + ); + // A POST SSE reply is bounded and goes down the body-capture path. + await fetcher("https://example.com/mcp", { method: "POST" }); + await flush(); + expect(updates).toEqual([]); + }); + + it("gives up quietly when the response cannot be cloned", async () => { + const response = new Response(controlledStream().stream, { + headers: { "content-type": "text/event-stream" }, + }); + vi.spyOn(response, "clone").mockImplementation(() => { + throw new TypeError("already consumed"); + }); + const updates: unknown[] = []; + const fetcher = createFetchTracker((async () => response) as typeof fetch, { + updateStream: (id, state) => updates.push({ id, state }), + }); + await expect( + fetcher("https://example.com/mcp", { method: "GET" }), + ).resolves.toBe(response); + await flush(); + expect(updates).toEqual([]); + }); + + it("gives up quietly when the clone has no body", async () => { + const response = new Response(controlledStream().stream, { + headers: { "content-type": "text/event-stream" }, + }); + vi.spyOn(response, "clone").mockImplementation( + () => + new Response(null, { + headers: { "content-type": "text/event-stream" }, + }), + ); + const updates: unknown[] = []; + const fetcher = createFetchTracker((async () => response) as typeof fetch, { + updateStream: (id, state) => updates.push({ id, state }), + }); + await fetcher("https://example.com/mcp", { method: "GET" }); + await flush(); + expect(updates).toEqual([]); + }); +}); diff --git a/clients/web/src/test/core/mcp/state/fetchRequestLogState.test.ts b/clients/web/src/test/core/mcp/state/fetchRequestLogState.test.ts index d8b239b3..5e00da33 100644 --- a/clients/web/src/test/core/mcp/state/fetchRequestLogState.test.ts +++ b/clients/web/src/test/core/mcp/state/fetchRequestLogState.test.ts @@ -146,6 +146,51 @@ describe("FetchRequestLogState", () => { expect(seen).toHaveLength(1); }); + it("patches the matching entry's stream state and re-emits on fetchRequestStreamUpdate (#2318)", () => { + client.dispatchTypedEvent("fetchRequest", entry("a")); + client.dispatchTypedEvent("fetchRequest", entry("b")); + const seen: FetchRequestEntry[][] = []; + state.addEventListener("fetchRequestsChange", (e) => seen.push(e.detail)); + + client.dispatchTypedEvent("fetchRequestStreamUpdate", { + id: "b", + stream: { eventCount: 2 }, + }); + const closedAt = new Date(2026, 4, 13, 0, 1); + client.dispatchTypedEvent("fetchRequestStreamUpdate", { + id: "b", + stream: { eventCount: 3, closedAt }, + }); + + const entries = state.getFetchRequests(); + expect(entries[0]?.stream).toBeUndefined(); + // Each update replaces the previous state wholesale. + expect(entries[1]?.stream).toEqual({ eventCount: 3, closedAt }); + expect(seen).toHaveLength(2); + }); + + it("ignores a stream update for an unknown id without emitting", () => { + client.dispatchTypedEvent("fetchRequest", entry("a")); + const seen: FetchRequestEntry[][] = []; + state.addEventListener("fetchRequestsChange", (e) => seen.push(e.detail)); + client.dispatchTypedEvent("fetchRequestStreamUpdate", { + id: "gone", + stream: { eventCount: 1 }, + }); + expect(seen).toHaveLength(0); + expect(state.getFetchRequests()[0]?.stream).toBeUndefined(); + }); + + it("stops patching stream state after destroy", () => { + client.dispatchTypedEvent("fetchRequest", entry("a")); + state.destroy(); + client.dispatchTypedEvent("fetchRequestStreamUpdate", { + id: "a", + stream: { eventCount: 1 }, + }); + expect(state.getFetchRequests()).toEqual([]); + }); + it("silently ignores a body update for an unknown id when the log is not full (benign straggler)", () => { // Below capacity, an idx === -1 means the entry was cleared or never // existed — not a rotation drop — so it must NOT warn or emit the dropped diff --git a/clients/web/src/test/core/react/useInspectorClient.test.tsx b/clients/web/src/test/core/react/useInspectorClient.test.tsx index 3f6c5793..613d329b 100644 --- a/clients/web/src/test/core/react/useInspectorClient.test.tsx +++ b/clients/web/src/test/core/react/useInspectorClient.test.tsx @@ -68,6 +68,42 @@ describe("useInspectorClient", () => { expect(result.current.instructions).toBe("after"); }); + it("returns the client's connection diagnostics and re-reads them on change (#2318)", () => { + const client = new FakeInspectorClient(); + const { result } = renderHook(() => useInspectorClient(client)); + expect(result.current.connectionDiagnostics).toEqual({ + capturedAt: 0, + outstandingRequests: [], + }); + const next = { + capturedAt: 10, + outstandingRequests: [{ id: 1, method: "tools/list", sentAt: 5 }], + lastResponse: { method: "initialize", receivedAt: 4 }, + }; + act(() => { + client.setConnectionDiagnostics(next); + }); + expect(result.current.connectionDiagnostics).toBe(next); + }); + + it("falls back to an empty snapshot for a client without the optional accessor", () => { + 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; + const { result } = renderHook(() => useInspectorClient(client)); + expect(result.current.connectionDiagnostics).toEqual({ + capturedAt: 0, + outstandingRequests: [], + }); + const { result: absent } = renderHook(() => useInspectorClient(null)); + expect(absent.current.connectionDiagnostics).toBe( + result.current.connectionDiagnostics, + ); + }); + it("subscribes to protocolVersionChange and updates", () => { const client = new FakeInspectorClient(); const { result } = renderHook(() => useInspectorClient(client)); 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 new file mode 100644 index 00000000..7e3b1172 --- /dev/null +++ b/clients/web/src/test/integration/mcp/inspectorClient-timeout-diagnostics.test.ts @@ -0,0 +1,351 @@ +import { describe, it, expect, afterEach } from "vitest"; +import { + createServer, + type IncomingMessage, + type Server, + type ServerResponse, +} from "node:http"; +import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; +import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js"; +import { FetchRequestLogState } from "@inspector/core/mcp/state/index.js"; +import { createTransportNode } from "@inspector/core/mcp/node/transport.js"; +import { eraToVersionNegotiation } from "@inspector/core/mcp/types.js"; +import type { + AnnotatedRequestTimeoutData, + ConnectionDiagnostics, +} from "@inspector/core/mcp/connectionDiagnostics.js"; + +/** + * Request-timeout diagnostics against a real connection (#2318). + * + * The #2187 shape: a server that answers `initialize`, holds the standalone + * `GET /mcp` notification stream open without ever emitting an event, and + * never answers `tools/list`. The SDK's per-request timeout fires and the + * Inspector's word on it used to be `Request timed out`. These tests drive the + * real transport, the real SDK client and the real timeout, and assert on what + * the rejection now says — plus the snapshot and the events it is built from. + * + * Hand-rolled rather than a composable test server because the fixture has + * to misbehave in ways the SDK's own server cannot be configured to: hold a + * request forever, and push raw bytes down the GET stream on cue. + */ + +/** A short budget so the timeout fires in test time rather than in 60s. */ +const REQUEST_TIMEOUT_MS = 300; + +type ToolsListMode = "hang" | "error"; + +interface HangingServer { + url: string; + /** Every JSON-RPC method the server was sent, in order. */ + calls: string[]; + /** Write one SSE event onto every open GET stream. */ + pushEvent: (message: Record) => void; + /** Write a comment-only keepalive block onto every open GET stream. */ + pushKeepalive: () => void; + /** How many GET streams have been opened so far. */ + streamsOpened: () => number; + setToolsListMode: (mode: ToolsListMode) => void; + stop: () => Promise; +} + +function startHangingServer(): Promise { + const calls: string[] = []; + const streams = new Set(); + const held = new Set(); + let streamsOpened = 0; + let toolsListMode: ToolsListMode = "hang"; + + const handler = async (req: IncomingMessage, res: ServerResponse) => { + if (req.method === "GET") { + streamsOpened += 1; + streams.add(res); + res.on("close", () => streams.delete(res)); + res.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-cache", + }); + // Headers out now, so the client's fetch resolves and the stream reads + // as open — without a byte of event data on it. + res.flushHeaders(); + return; + } + if (req.method === "DELETE") { + res.writeHead(200).end(); + return; + } + const chunks: Buffer[] = []; + for await (const chunk of req) chunks.push(chunk as Buffer); + const body = JSON.parse(Buffer.concat(chunks).toString()) as { + id?: unknown; + method: string; + }; + calls.push(body.method); + // Notifications (`notifications/initialized`, and the SDK's own + // `notifications/cancelled` when the timeout fires) get an empty 202. + if (body.id === undefined) { + res.writeHead(202).end(); + return; + } + const reply = (payload: Record) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ jsonrpc: "2.0", id: body.id, ...payload })); + }; + if (body.method === "initialize") { + reply({ + result: { + protocolVersion: "2025-06-18", + capabilities: { tools: {} }, + serverInfo: { name: "hanging-server", version: "1.0.0" }, + }, + }); + return; + } + if (body.method === "tools/list") { + if (toolsListMode === "error") { + reply({ error: { code: -32603, message: "tools/list exploded" } }); + return; + } + // Never answered. Kept so teardown can release the socket. + held.add(res); + res.on("close", () => held.delete(res)); + return; + } + reply({ result: {} }); + }; + + const server: Server = createServer((req, res) => { + void handler(req, res); + }); + + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + /* v8 ignore next -- listen() on a fresh server always yields an AddressInfo */ + const port = typeof address === "object" && address ? address.port : 0; + resolve({ + url: `http://127.0.0.1:${port}/mcp`, + calls, + pushEvent: (message) => { + for (const res of streams) { + res.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`); + } + }, + pushKeepalive: () => { + for (const res of streams) res.write(": keepalive\n\n"); + }, + streamsOpened: () => streamsOpened, + setToolsListMode: (mode) => { + toolsListMode = mode; + }, + stop: () => + new Promise((done) => { + for (const res of held) res.destroy(); + for (const res of streams) res.destroy(); + server.closeAllConnections(); + server.close(() => done()); + }), + }); + }); + }); +} + +/** Poll until `predicate` holds; the failure names what never became true. */ +async function waitUntil( + what: string, + predicate: () => boolean, + timeoutMs = 5000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > deadline) { + throw new Error(`timed out waiting for ${what}`); + } + await new Promise((r) => setTimeout(r, 10)); + } +} + +describe("InspectorClient request-timeout diagnostics (#2318)", () => { + let client: InspectorClient | null = null; + let server: HangingServer | null = null; + + async function connectTo(url: string): Promise { + client = new InspectorClient( + { type: "streamable-http", url }, + { + environment: { transport: createTransportNode }, + // Pinned legacy so the connect is exactly `initialize` and nothing + // else — the assertions below name the last response by method. + versionNegotiation: eraToVersionNegotiation("legacy"), + timeout: REQUEST_TIMEOUT_MS, + }, + ); + await client.connect(); + return client; + } + + /** The SDK opens the GET stream after `notifications/initialized`, off the connect path. */ + async function waitForNotificationStream(c: InspectorClient): Promise { + await waitUntil( + "the notification stream to be tracked", + () => c.getConnectionDiagnostics().notificationStream !== undefined, + ); + } + + afterEach(async () => { + try { + await client?.disconnect(); + } catch { + // Teardown only — the assertions already ran. + } + client = null; + await server?.stop(); + server = null; + }); + + it("annotates the timeout with the request, the unanswered set, the last response and the stream", async () => { + server = await startHangingServer(); + const c = await connectTo(server.url); + await waitForNotificationStream(c); + server.pushKeepalive(); + + let caught: unknown; + try { + await c.listTools(); + } catch (err) { + caught = err; + } + + expect(SdkError.isInstance(caught)).toBe(true); + const error = caught as SdkError; + expect(error.code).toBe(SdkErrorCode.RequestTimeout); + expect(error.message).toMatch( + new RegExp( + "^Request timed out after 300ms \\(tools/list\\)\\. " + + "1 request is unanswered: tools/list \\(sent \\d+(ms|s) ago\\)\\. " + + "Last response received \\d+(ms|s) ago \\(initialize\\)\\. " + + "Notification stream \\(GET /mcp\\) open for \\d+(ms|s), 0 events delivered\\.$", + ), + ); + // The structured form rides along for consumers that want it. + const data = error.data as AnnotatedRequestTimeoutData; + expect(data.timeout).toBe(REQUEST_TIMEOUT_MS); + expect(data.method).toBe("tools/list"); + expect(data.diagnostics.outstandingRequests.map((r) => r.method)).toEqual([ + "tools/list", + ]); + expect(data.diagnostics.lastResponse?.method).toBe("initialize"); + expect(data.diagnostics.notificationStream?.eventCount).toBe(0); + // The SDK gave up on the request; the connection did not answer it. It is + // still unanswered, and the live snapshot keeps saying so. + expect( + c.getConnectionDiagnostics().outstandingRequests.map((r) => r.method), + ).toEqual(["tools/list"]); + // The keepalive comment reached the client and was not counted. + expect(c.getConnectionDiagnostics().notificationStream?.eventCount).toBe(0); + // The SDK's legacy cancel rode the wire and the server saw the whole + // exchange in order. The cancel is fire-and-forget on the SDK side — the + // rejection lands before the POST does — so wait for it to arrive. + const calls = server.calls; + await waitUntil("the cancellation to reach the server", () => + calls.includes("notifications/cancelled"), + ); + expect(calls).toEqual([ + "initialize", + "notifications/initialized", + "tools/list", + "notifications/cancelled", + ]); + }); + + it("emits connectionDiagnosticsChange as the state moves, and counts stream events", async () => { + server = await startHangingServer(); + const c = await connectTo(server.url); + const log = new FetchRequestLogState(c); + const snapshots: ConnectionDiagnostics[] = []; + c.addEventListener("connectionDiagnosticsChange", (event) => + snapshots.push(event.detail), + ); + try { + await waitForNotificationStream(c); + const stream = c.getConnectionDiagnostics().notificationStream!; + expect(stream.url).toBe(server.url); + expect(stream.closedAt).toBeUndefined(); + + // A server push on the notification stream is an event. + server.pushEvent({ + jsonrpc: "2.0", + method: "notifications/tools/list_changed", + }); + await waitUntil( + "the stream event to be counted", + () => c.getConnectionDiagnostics().notificationStream?.eventCount === 1, + ); + // ...and the Network log's entry for the GET carries the same state. + const getEntry = log + .getFetchRequests() + .find((entry) => entry.method === "GET"); + expect(getEntry?.stream).toEqual({ eventCount: 1 }); + + // The timeout path dispatched a snapshot with the request outstanding. + await expect(c.listTools()).rejects.toThrow(/Request timed out/); + expect( + snapshots.some((s) => + s.outstandingRequests.some((r) => r.method === "tools/list"), + ), + ).toBe(true); + + // Disconnecting aborts the stream: the watcher reports the close and + // the snapshot — which outlives the session until the next connect — + // records it. + await c.disconnect(); + await waitUntil( + "the stream close to be recorded", + () => + c.getConnectionDiagnostics().notificationStream?.closedAt !== + undefined, + ); + expect(c.getConnectionDiagnostics().notificationStream).toMatchObject({ + eventCount: 1, + }); + } finally { + log.destroy(); + } + }); + + it("starts the next session clean", async () => { + server = await startHangingServer(); + const c = await connectTo(server.url); + await waitForNotificationStream(c); + await expect(c.listTools()).rejects.toThrow(/Request timed out/); + expect(c.getConnectionDiagnostics().outstandingRequests).toHaveLength(1); + + await c.disconnect(); + await c.connect(); + + const fresh = c.getConnectionDiagnostics(); + expect(fresh.outstandingRequests).toEqual([]); + expect(fresh.lastResponse?.method).toBe("initialize"); + // Whether the new session's GET has landed yet is timing; what must not be + // there is the previous session's stream. + await waitForNotificationStream(c); + expect(c.getConnectionDiagnostics().notificationStream?.closedAt).toBe( + undefined, + ); + expect(server.streamsOpened()).toBe(2); + }); + + it("leaves a failure that is not a timeout exactly as the SDK threw it", async () => { + server = await startHangingServer(); + server.setToolsListMode("error"); + const c = await connectTo(server.url); + + await expect(c.listTools()).rejects.toThrow(/tools\/list exploded/); + await expect(c.listTools()).rejects.not.toThrow(/unanswered/); + // Answered (with an error), so nothing is outstanding. + expect(c.getConnectionDiagnostics().outstandingRequests).toEqual([]); + expect(c.getConnectionDiagnostics().lastResponse?.method).toBe( + "tools/list", + ); + }); +}); diff --git a/clients/web/src/test/integration/mcp/remote/remote-session.test.ts b/clients/web/src/test/integration/mcp/remote/remote-session.test.ts index c152b960..d11aee37 100644 --- a/clients/web/src/test/integration/mcp/remote/remote-session.test.ts +++ b/clients/web/src/test/integration/mcp/remote/remote-session.test.ts @@ -194,6 +194,30 @@ describe("RemoteSession", () => { }); }); + it("onFetchStreamUpdate forwards the event count, serializing closedAt to ISO when present (#2318)", () => { + const session = new RemoteSession("s10a"); + const received: { type: string; data: unknown }[] = []; + session.setEventConsumer((event) => + received.push({ type: event.type, data: event.data }), + ); + session.onFetchStreamUpdate("req-1", { eventCount: 2 }); + session.onFetchStreamUpdate("req-1", { + eventCount: 3, + closedAt: new Date("2026-01-01T00:00:05Z"), + }); + expect(received).toEqual([ + { type: "fetch_stream_update", data: { id: "req-1", eventCount: 2 } }, + { + type: "fetch_stream_update", + data: { + id: "req-1", + eventCount: 3, + closedAt: "2026-01-01T00:00:05.000Z", + }, + }, + ]); + }); + it("onFetchRequest serializes a Date timestamp to an ISO string", () => { const session = new RemoteSession("s11"); const received: { type: string; data: unknown }[] = []; 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 fddc10c9..ef7ebf5f 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 @@ -340,6 +340,59 @@ describe("RemoteClientTransport (focused branch coverage)", () => { expect(onFetchResponseBody).toHaveBeenCalledWith("f1", "BODY"); }); + it("delivers fetch_stream_update events, rebuilding closedAt as a Date (#2318)", async () => { + const onFetchStreamUpdate = vi.fn(); + const t = makeTransport( + { + events: () => + sseResponse([ + sseFrame({ + type: "fetch_stream_update", + data: { id: "f1", eventCount: 2 }, + }), + sseFrame({ + type: "fetch_stream_update", + data: { + id: "f1", + eventCount: 3, + closedAt: "2026-01-01T00:00:05.000Z", + }, + }), + ]), + }, + { onFetchStreamUpdate }, + ); + await t.start(); + await tick(); + expect(onFetchStreamUpdate).toHaveBeenCalledTimes(2); + expect(onFetchStreamUpdate).toHaveBeenNthCalledWith(1, "f1", { + eventCount: 2, + }); + expect(onFetchStreamUpdate).toHaveBeenNthCalledWith(2, "f1", { + eventCount: 3, + closedAt: new Date("2026-01-01T00:00:05.000Z"), + }); + }); + + it("drops fetch_stream_update events when no handler is registered", async () => { + const onFetchRequest = vi.fn(); + const t = makeTransport( + { + events: () => + sseResponse([ + sseFrame({ + type: "fetch_stream_update", + data: { id: "f1", eventCount: 1 }, + }), + ]), + }, + { onFetchRequest }, + ); + await t.start(); + await tick(); + expect(onFetchRequest).not.toHaveBeenCalled(); + }); + it("delivers stdio_log events to onStderr", async () => { const onStderr = vi.fn(); const t = makeTransport( diff --git a/clients/web/src/utils/connectionActivity.test.ts b/clients/web/src/utils/connectionActivity.test.ts new file mode 100644 index 00000000..db31480e --- /dev/null +++ b/clients/web/src/utils/connectionActivity.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it } from "vitest"; +import { + formatLastResponse, + formatNotificationStream, + formatOutstandingRequests, + NO_NOTIFICATION_STREAM_LABEL, + NO_RESPONSE_YET_LABEL, + NO_STREAM_ON_STDIO_LABEL, +} from "./connectionActivity"; + +const NOW = 1_000_000; +const seconds = (n: number) => n * 1000; + +describe("formatOutstandingRequests", () => { + it("renders one line per request, oldest first, against the given clock", () => { + expect( + formatOutstandingRequests( + { + capturedAt: NOW, + outstandingRequests: [ + { id: 1, method: "tools/list", sentAt: NOW - seconds(60) }, + { id: 2, method: "ping", sentAt: NOW - seconds(12) }, + ], + }, + NOW, + ), + ).toEqual(["tools/list — sent 1m00s ago", "ping — sent 12s ago"]); + }); + + it("renders nothing when nothing is outstanding", () => { + expect( + formatOutstandingRequests( + { capturedAt: NOW, outstandingRequests: [] }, + NOW, + ), + ).toEqual([]); + }); +}); + +describe("formatLastResponse", () => { + it("names the method and the age", () => { + expect( + formatLastResponse( + { + capturedAt: NOW, + outstandingRequests: [], + lastResponse: { method: "initialize", receivedAt: NOW - 500 }, + }, + NOW, + ), + ).toBe("initialize — 500ms ago"); + }); + + it("says none yet before any response", () => { + expect( + formatLastResponse({ capturedAt: NOW, outstandingRequests: [] }, NOW), + ).toBe(NO_RESPONSE_YET_LABEL); + }); +}); + +describe("the clock argument", () => { + it("moves every duration with it, so a ticking caller renders live values", () => { + const diagnostics = { + capturedAt: NOW, + outstandingRequests: [ + { id: 1, method: "tools/list", sentAt: NOW - seconds(5) }, + ], + lastResponse: { method: "initialize", receivedAt: NOW - seconds(6) }, + }; + expect(formatOutstandingRequests(diagnostics, NOW + seconds(55))).toEqual([ + "tools/list — sent 1m00s ago", + ]); + expect(formatLastResponse(diagnostics, NOW + seconds(55))).toBe( + "initialize — 1m01s ago", + ); + }); +}); + +describe("formatNotificationStream", () => { + const open = { + url: "http://h/mcp", + openedAt: NOW - seconds(252), + eventCount: 1, + }; + + it("is not applicable on stdio, whatever the snapshot says", () => { + expect(formatNotificationStream(open, "stdio", NOW)).toBe( + NO_STREAM_ON_STDIO_LABEL, + ); + }); + + it("says not opened when no stream has been tracked", () => { + expect(formatNotificationStream(undefined, "sse", NOW)).toBe( + NO_NOTIFICATION_STREAM_LABEL, + ); + }); + + it("describes an open stream by endpoint, age and event count", () => { + expect(formatNotificationStream(open, "streamable-http", NOW)).toBe( + "GET /mcp — open for 4m12s, 1 event delivered", + ); + }); + + it("describes a closed stream by when it closed and how long it lasted", () => { + expect( + formatNotificationStream( + { ...open, eventCount: 2, closedAt: NOW - seconds(5) }, + "streamable-http", + NOW, + ), + ).toBe("GET /mcp — closed 5s ago after 4m07s, 2 events delivered"); + }); +}); diff --git a/clients/web/src/utils/connectionActivity.ts b/clients/web/src/utils/connectionActivity.ts new file mode 100644 index 00000000..649785f4 --- /dev/null +++ b/clients/web/src/utils/connectionActivity.ts @@ -0,0 +1,63 @@ +// Row formatters for the Connection Info modal's Connection Activity section +// (#2318): the same snapshot a request timeout's message is written from, +// rendered while the request is still in flight. Pure functions over the +// snapshot and an explicit clock — the caller decides what "now" is (the +// snapshot's own `capturedAt` on first paint, a ticking clock afterwards; see +// `useTickingClock`), so nothing here reads the wall clock and a render stays +// a pure function of its inputs. + +import { + describeStreamEndpoint, + formatDuration, + type ConnectionDiagnostics, + type NotificationStreamState, +} from "@inspector/core/mcp/connectionDiagnostics.js"; +import type { ServerType } from "@inspector/core/mcp/types.js"; + +export const NO_OUTSTANDING_REQUESTS_LABEL = "None"; +export const NO_RESPONSE_YET_LABEL = "None this session"; +export const NO_NOTIFICATION_STREAM_LABEL = "Not opened"; +export const NO_STREAM_ON_STDIO_LABEL = "N/A (stdio)"; + +/** + * One line per unanswered request, oldest first — `tools/list — sent 60s + * ago`. A request the SDK has already given up on stays listed: it is still + * unanswered, which is the fact this section reports. + */ +export function formatOutstandingRequests( + diagnostics: ConnectionDiagnostics, + now: number, +): string[] { + return diagnostics.outstandingRequests.map( + (request) => + `${request.method} — sent ${formatDuration(now - request.sentAt)} ago`, + ); +} + +export function formatLastResponse( + diagnostics: ConnectionDiagnostics, + now: number, +): string { + const last = diagnostics.lastResponse; + if (!last) return NO_RESPONSE_YET_LABEL; + return `${last.method} — ${formatDuration(now - last.receivedAt)} ago`; +} + +/** + * The notification stream row. stdio has no HTTP stream at all, so the row + * says so rather than "Not opened", which would read as a fault. + */ +export function formatNotificationStream( + stream: NotificationStreamState | undefined, + transport: ServerType, + now: number, +): string { + if (transport === "stdio") return NO_STREAM_ON_STDIO_LABEL; + if (!stream) return NO_NOTIFICATION_STREAM_LABEL; + const endpoint = describeStreamEndpoint(stream.url); + const events = `${stream.eventCount} event${stream.eventCount === 1 ? "" : "s"} delivered`; + if (stream.closedAt !== undefined) { + return `${endpoint} — closed ${formatDuration(now - stream.closedAt)} ago after ${formatDuration(stream.closedAt - stream.openedAt)}, ${events}`; + } + return `${endpoint} — open for ${formatDuration(now - stream.openedAt)}, ${events}`; +} diff --git a/clients/web/src/utils/uncapturedBodyNote.test.ts b/clients/web/src/utils/uncapturedBodyNote.test.ts new file mode 100644 index 00000000..d907286d --- /dev/null +++ b/clients/web/src/utils/uncapturedBodyNote.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from "vitest"; +import type { FetchRequestEntry } from "@inspector/core/mcp/types.js"; +import { + isLongLivedStreamEntry, + uncapturedBodyNote, +} from "./uncapturedBodyNote"; + +const streamEntry: FetchRequestEntry = { + id: "n-1", + timestamp: new Date("2026-03-17T10:00:00Z"), + method: "GET", + url: "https://example.com/mcp", + requestHeaders: {}, + responseStatus: 200, + responseHeaders: { "content-type": "text/event-stream" }, + category: "transport", +}; + +describe("isLongLivedStreamEntry", () => { + it("is true for a GET answered with an event stream and false otherwise", () => { + expect(isLongLivedStreamEntry(streamEntry)).toBe(true); + expect(isLongLivedStreamEntry({ ...streamEntry, method: "POST" })).toBe( + false, + ); + expect( + isLongLivedStreamEntry({ ...streamEntry, responseHeaders: undefined }), + ).toBe(false); + }); +}); + +describe("uncapturedBodyNote", () => { + it("says (empty) for a bounded response", () => { + expect(uncapturedBodyNote({ ...streamEntry, method: "POST" })).toBe( + "(empty)", + ); + }); + + it("says only that the body was not captured before the stream is watched", () => { + expect(uncapturedBodyNote(streamEntry)).toBe( + "Long-lived stream — body not captured", + ); + }); + + it("reports the event count and open state, with the singular for one event", () => { + expect( + uncapturedBodyNote({ ...streamEntry, stream: { eventCount: 3 } }), + ).toBe( + "Long-lived stream — 3 events delivered, still open; body not captured", + ); + expect( + uncapturedBodyNote({ + ...streamEntry, + stream: { eventCount: 1, closedAt: new Date() }, + }), + ).toBe("Long-lived stream — 1 event delivered, closed; body not captured"); + }); +}); diff --git a/clients/web/src/utils/uncapturedBodyNote.ts b/clients/web/src/utils/uncapturedBodyNote.ts new file mode 100644 index 00000000..636c7f33 --- /dev/null +++ b/clients/web/src/utils/uncapturedBodyNote.ts @@ -0,0 +1,29 @@ +// 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 type { FetchRequestEntry } from "@inspector/core/mcp/types.js"; + +/** Whether the entry is an unbounded server-to-client stream (never buffered). */ +export function isLongLivedStreamEntry(entry: FetchRequestEntry): boolean { + return isLongLivedStreamResponse( + entry.method, + entry.responseHeaders?.["content-type"], + ); +} + +/** + * 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 + * watching it: how many events it delivered and whether it is still open. + * Before the first update, or for a log restored from a session that + * predates the field, the note says only that the body was not captured. + */ +export function uncapturedBodyNote(entry: FetchRequestEntry): string { + if (!isLongLivedStreamEntry(entry)) return "(empty)"; + const stream = entry.stream; + if (!stream) return "Long-lived stream — body not captured"; + const events = `${stream.eventCount} event${stream.eventCount === 1 ? "" : "s"} delivered`; + const state = stream.closedAt ? "closed" : "still open"; + return `Long-lived stream — ${events}, ${state}; body not captured`; +} diff --git a/core/mcp/__tests__/fakeInspectorClient.ts b/core/mcp/__tests__/fakeInspectorClient.ts index bf77d19c..41bcfcc1 100644 --- a/core/mcp/__tests__/fakeInspectorClient.ts +++ b/core/mcp/__tests__/fakeInspectorClient.ts @@ -40,6 +40,7 @@ import type { } from "../types.js"; import { INACTIVE_SUBSCRIPTION_STREAM_STATE } from "../types.js"; import type { MalformedListItem } from "../listSalvage.js"; +import type { ConnectionDiagnostics } from "../connectionDiagnostics.js"; import type { SkillEntry, SkillResource } from "../skillsSchemas.js"; import type { SkillsExtensionSupport } from "../skills.js"; import type { JsonValue } from "../../json/jsonUtils.js"; @@ -321,6 +322,21 @@ export class FakeInspectorClient return this.discoverResult; } + private connectionDiagnostics: ConnectionDiagnostics = { + capturedAt: 0, + outstandingRequests: [], + }; + + getConnectionDiagnostics(): ConnectionDiagnostics { + return this.connectionDiagnostics; + } + + /** Test helper: set the diagnostics snapshot and emit the change (#2318). */ + setConnectionDiagnostics(diagnostics: ConnectionDiagnostics): void { + this.connectionDiagnostics = diagnostics; + this.dispatchTypedEvent("connectionDiagnosticsChange", diagnostics); + } + resourceSubscriptionStreamState: ResourceSubscriptionStreamState = INACTIVE_SUBSCRIPTION_STREAM_STATE; diff --git a/core/mcp/connectionDiagnostics.ts b/core/mcp/connectionDiagnostics.ts new file mode 100644 index 00000000..9163f557 --- /dev/null +++ b/core/mcp/connectionDiagnostics.ts @@ -0,0 +1,252 @@ +/** + * Connection diagnostics: what the client is still waiting on, when it last + * heard back, and the state of the server-to-client notification stream — and + * the one place that turns the SDK's bare `Request timed out` into a message + * that reports those facts (#2318). + * + * The motivating case is #2187: a server that serves one request at a time. + * The standalone `GET /mcp` stream the client opens after `initialize` + * occupies that single slot, so `initialize` succeeds and every request after + * it hangs — and the Inspector's only word on it was `Request timed out`, + * which names the symptom the user can already see. Every fact needed to + * suspect the cause was already in-process at that moment: the stream had been + * open for the whole session and delivered nothing, `tools/list` had been + * outstanding for 60s, and nothing had been answered since `initialize`. + * + * Everything here is pure. `InspectorClient` owns the bookkeeping (it sees + * every outbound request, every inbound response and every tracked fetch) and + * hands a snapshot to {@link annotateRequestTimeout}; the UI clients render the + * same snapshot through `getConnectionDiagnostics()`. + * + * Deliberately factual: the summary reports what the connection's state *is* + * and lets the reader draw the conclusion. It does not assert a diagnosis the + * Inspector cannot verify from its own side. + */ + +import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client"; + +/** A request this client sent that no response has arrived for. */ +export interface OutstandingRequest { + id: string | number; + method: string; + /** Epoch ms at which the request went out. */ + sentAt: number; +} + +/** The most recent response received to one of this client's requests. */ +export interface LastResponse { + /** The method of the request it answered. */ + method: string; + /** Epoch ms at which it arrived. */ + receivedAt: number; +} + +/** + * The most recent long-lived server-to-client stream the transport opened — + * the standalone `GET` on Streamable HTTP, or the primary event stream on + * legacy SSE. Absent on stdio, and until the first such stream is opened. + */ +export interface NotificationStreamState { + /** The request URL, as recorded (query already redacted). */ + url: string; + /** Epoch ms at which the response headers arrived and the stream opened. */ + openedAt: number; + /** SSE events delivered so far — messages, not keepalive comments. */ + eventCount: number; + /** Epoch ms at which the stream ended, once it has. */ + closedAt?: number; +} + +export interface ConnectionDiagnostics { + /** + * Epoch ms at which this snapshot was taken. Every duration derived from + * the snapshot ("sent 60s ago", "open for 4m12s") is measured against this + * clock rather than against the moment of rendering, so the snapshot is + * self-contained: a message written from it, a component rendering it and + * a test asserting on it all read the same numbers. + */ + capturedAt: number; + /** + * Every request sent this session that has received no response, oldest + * first. A request the SDK gave up on (a timeout, a cancel) stays here: it + * is still unanswered, which is the fact being reported. + */ + outstandingRequests: OutstandingRequest[]; + lastResponse?: LastResponse; + notificationStream?: NotificationStreamState; +} + +/** + * The data the SDK attaches to a per-request timeout: `{ timeout }` in ms. + * Narrowed here rather than imported because the SDK types it `unknown`. + */ +interface RequestTimeoutData { + timeout: number; +} + +function isRequestTimeoutData(value: unknown): value is RequestTimeoutData { + return ( + typeof value === "object" && + value !== null && + typeof (value as { timeout?: unknown }).timeout === "number" + ); +} + +/** + * Whether `err` is the SDK's per-request timeout: an `SdkError` with + * `RequestTimeout` and a numeric `timeout` in its data. + * + * The data check is not decoration. `Protocol.request` reuses the + * `RequestTimeout` code for two other rejections that are not timeouts at + * all — an abort through `options.signal` whose reason is not an `SdkError` + * (the Cancel button), and "Maximum total timeout exceeded" — and neither + * carries `{ timeout }`. Only the genuine per-request timeout does. + */ +export function isRequestTimeoutError( + err: unknown, +): err is SdkError & { data: RequestTimeoutData } { + return ( + SdkError.isInstance(err) && + err.code === SdkErrorCode.RequestTimeout && + isRequestTimeoutData(err.data) + ); +} + +/** + * A compact human duration: `850ms`, `12s`, `4m12s`, `1h02m`. Sub-second + * values are shown in ms so a very short timeout still reads as a number + * rather than as `0s`. + */ +export function formatDuration(ms: number): string { + const clamped = Math.max(0, ms); + if (clamped < 1000) return `${Math.round(clamped)}ms`; + const totalSeconds = Math.round(clamped / 1000); + const hours = Math.floor(totalSeconds / 3600); + const minutes = Math.floor((totalSeconds % 3600) / 60); + const seconds = totalSeconds % 60; + if (hours > 0) return `${hours}h${String(minutes).padStart(2, "0")}m`; + if (minutes > 0) return `${minutes}m${String(seconds).padStart(2, "0")}s`; + return `${seconds}s`; +} + +/** + * How many outstanding requests the summary names individually before + * collapsing the rest into "and N more". Enough to show the shape of a stall + * (the one everything is queued behind, plus what queued behind it) without + * turning a toast into a page. + */ +export const MAX_LISTED_OUTSTANDING_REQUESTS = 5; + +/** + * `GET /mcp` for a stream URL: method plus path only. The origin is the + * server the user is already connected to, and the query string (already + * redacted at tracking time) adds nothing to a one-line summary. Falls back to + * the raw string for an unparseable URL rather than dropping the clause. + */ +export function describeStreamEndpoint(url: string): string { + try { + return `GET ${new URL(url).pathname}`; + } catch { + return `GET ${url}`; + } +} + +function plural(count: number, noun: string): string { + return `${count} ${noun}${count === 1 ? "" : "s"}`; +} + +function describeOutstanding( + requests: OutstandingRequest[], + now: number, +): string { + if (requests.length === 0) return "No requests are unanswered."; + const listed = requests + .slice(0, MAX_LISTED_OUTSTANDING_REQUESTS) + .map((r) => `${r.method} (sent ${formatDuration(now - r.sentAt)} ago)`); + const rest = requests.length - listed.length; + const tail = rest > 0 ? `, and ${rest} more` : ""; + const verb = requests.length === 1 ? "is" : "are"; + return `${plural(requests.length, "request")} ${verb} unanswered: ${listed.join(", ")}${tail}.`; +} + +function describeLastResponse( + lastResponse: LastResponse | undefined, + now: number, +): string { + if (!lastResponse) return "No response has been received this session."; + return `Last response received ${formatDuration(now - lastResponse.receivedAt)} ago (${lastResponse.method}).`; +} + +function describeStream( + stream: NotificationStreamState | undefined, + now: number, +): string | undefined { + if (!stream) return undefined; + const endpoint = describeStreamEndpoint(stream.url); + const events = plural(stream.eventCount, "event"); + if (stream.closedAt !== undefined) { + return `Notification stream (${endpoint}) closed ${formatDuration(now - stream.closedAt)} ago after ${formatDuration(stream.closedAt - stream.openedAt)}, ${events} delivered.`; + } + return `Notification stream (${endpoint}) open for ${formatDuration(now - stream.openedAt)}, ${events} delivered.`; +} + +/** + * One paragraph describing the connection's state as of the snapshot's own + * clock: outstanding requests, the last response, and the notification + * stream when there is one. Each clause is a sentence, so a consumer that + * wants only one of them can split on the sentence boundary rather than + * re-deriving it. + */ +export function describeConnectionDiagnostics( + diagnostics: ConnectionDiagnostics, +): string { + const now = diagnostics.capturedAt; + const parts = [ + describeOutstanding(diagnostics.outstandingRequests, now), + describeLastResponse(diagnostics.lastResponse, now), + describeStream(diagnostics.notificationStream, now), + ]; + return parts.filter((p): p is string => p !== undefined).join(" "); +} + +/** + * The data carried by an annotated timeout: the SDK's own `{ timeout }`, plus + * the method that timed out and the snapshot the message was written from. + * Exposed as a type so a consumer that wants the structured form (a details + * modal, a CLI's JSON output) can read it off `error.data` instead of parsing + * the message. + */ +export interface AnnotatedRequestTimeoutData extends RequestTimeoutData { + method: string; + diagnostics: ConnectionDiagnostics; +} + +/** + * Rebuild the SDK's per-request timeout with a message that says which + * request timed out and what the connection looked like at that moment. + * + * Anything that is not that exact error is returned untouched — the SDK's + * other `RequestTimeout` rejections (see {@link isRequestTimeoutError}), and + * every other failure — so this is safe to apply blindly on the rejection + * path of every request. + * + * The result is a fresh `SdkError` with the same code, so an `isInstance` or + * code check downstream sees exactly what it saw before; only the message and + * the data change. The original is kept as `cause`. + */ +export function annotateRequestTimeout( + err: unknown, + method: string, + diagnostics: ConnectionDiagnostics, +): unknown { + if (!isRequestTimeoutError(err)) return err; + const data: AnnotatedRequestTimeoutData = { + ...err.data, + method, + diagnostics, + }; + const message = `Request timed out after ${formatDuration(err.data.timeout)} (${method}). ${describeConnectionDiagnostics(diagnostics)}`; + const annotated = new SdkError(SdkErrorCode.RequestTimeout, message, data); + annotated.cause = err; + return annotated; +} diff --git a/core/mcp/fetchTracking.ts b/core/mcp/fetchTracking.ts index 7e1e5864..fa4b63bd 100644 --- a/core/mcp/fetchTracking.ts +++ b/core/mcp/fetchTracking.ts @@ -1,4 +1,4 @@ -import type { FetchRequestEntryBase } from "./types.js"; +import type { FetchRequestEntryBase, FetchStreamState } from "./types.js"; /** * Header names whose values are replaced with `REDACTED_HEADER_VALUE` before a @@ -190,11 +190,18 @@ export function redactBody( return body; } -/** Case-insensitive lookup of a header value from a plain header record. */ -function findHeader( - headers: Record, +/** + * Case-insensitive lookup of a header value from a plain header record. + * Exported because a recorded entry's header casing is host-dependent (Node + * lowercases, happy-dom preserves the wire casing), so any consumer reading + * one back — the Network UI, the connection diagnostics — needs the same + * tolerance the tracker itself has. + */ +export function findHeader( + headers: Record | undefined, name: string, ): string | undefined { + if (!headers) return undefined; const target = name.toLowerCase(); for (const [key, value] of Object.entries(headers)) { if (key.toLowerCase() === target) return value; @@ -233,6 +240,98 @@ export interface FetchTrackingCallbacks { * never invoked and the entry's responseBody stays undefined. */ updateResponseBody?: (id: string, responseBody: string) => void; + /** + * Called for a long-lived stream response — the one case `updateResponseBody` + * never covers — each time an SSE event is delivered on it, and once more + * when it ends (with `closedAt` set). Lets the consumer show that the stream + * is open and what it has carried, which is the whole diagnostic when a + * server serializes requests behind it (#2318). + */ + updateStream?: (id: string, stream: FetchStreamState) => void; +} + +/** + * Count the SSE 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. + * + * 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 + * network error, or the transport aborting the fetch on disconnect (which + * errors both tee branches) — with `closedAt` set. The reader is never cancelled from here: + * cancelling one branch of a tee does not release the source, and + * `ReadableStream.cancel()` adopts the source's promise, which on a teardown + * path may never settle (the same hazard `core/mcp/remote/node/server.ts` + * documents at its own `cancel()` call). + */ +function watchLongLivedStream( + response: Response, + id: string, + callbacks: FetchTrackingCallbacks, +): void { + const update = callbacks.updateStream; + if (!update) return; + let body: ReadableStream | null; + try { + body = response.clone().body; + } catch { + // Clone failed (consumed body, transport quirks) — there is nothing to + // watch; the entry stays as tracked with no stream state. + return; + } + if (!body) return; + const reader = body.getReader(); + const decoder = new TextDecoder(); + let eventCount = 0; + let pending = ""; + let blockHasData = false; + // 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. + update(id, { eventCount }); + + const consumeLine = (rawLine: string): void => { + const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + if (line === "") { + if (blockHasData) { + eventCount += 1; + blockHasData = false; + update(id, { eventCount }); + } + return; + } + // A field line is `name[:value]`; `data` with no colon is still a data + // field per the spec, so match the name rather than the `data:` prefix. + if (line === "data" || line.startsWith("data:")) blockHasData = true; + }; + + 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"); + } + } + } catch { + // An errored or aborted stream has still ended — fall through to report + // the close; the entry's `error` field is for a fetch that never got a + // response, which this one did. + } + update(id, { eventCount, closedAt: new Date() }); + }; + void pump(); } /** @@ -378,6 +477,13 @@ export function createFetchTracker( callbacks.trackRequest?.(entry); + // A long-lived stream is watched instead of read: its body never ends, + // 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); + } + // Kick off a fire-and-forget read of the cloned body. The clone is an // independent tee'd stream so the transport keeps consuming the // original at its own pace. When the read resolves we patch the entry diff --git a/core/mcp/inspectorClient.ts b/core/mcp/inspectorClient.ts index 91cc7ff8..95b46f33 100644 --- a/core/mcp/inspectorClient.ts +++ b/core/mcp/inspectorClient.ts @@ -13,6 +13,7 @@ import type { MessageOrigin, FetchRequestEntry, FetchRequestEntryBase, + FetchStreamState, InspectorServerSettings, ResourceReadInvocation, ResourceTemplateReadInvocation, @@ -264,10 +265,29 @@ import { import type { TokenRevocationOutcome } from "../auth/revocation.js"; import type { OAuthTokens } from "@modelcontextprotocol/client"; import { silentLogger, type InspectorLogger } from "../logging/logger.js"; -import { createFetchTracker } from "./fetchTracking.js"; +import { + createFetchTracker, + findHeader, + isLongLivedStreamResponse, +} from "./fetchTracking.js"; +import { + annotateRequestTimeout, + type ConnectionDiagnostics, + type LastResponse, + type NotificationStreamState, + type OutstandingRequest, +} from "./connectionDiagnostics.js"; import { OAuthManager, type OAuthManagerConfig } from "./oauthManager.js"; import { RemoteClientTransport } from "./remote/remoteClientTransport.js"; +/** + * The notification stream the diagnostics are following, plus the fetch entry + * id that ties later `fetchRequestStreamUpdate`s back to it (#2318). + */ +interface TrackedNotificationStream extends NotificationStreamState { + id: string; +} + /** Internal record for a receiver task (server polls us for status/result). */ interface ReceiverTaskRecord { task: Task; @@ -475,13 +495,23 @@ export class InspectorClient extends InspectorClientEventTarget { private outputValidator: AjvJsonSchemaValidator | null = null; private transport: Transport | MessageTrackingTransport | null = null; private baseTransport: Transport | null = null; - // Correlation for `markResponseRejected` (#1953): the method of each - // outbound request still awaiting a response, and — once one is answered — - // the id of the most recently answered request per method. Entries are - // dropped as responses arrive, so this holds at most one id per method - // rather than growing with the session. - private outboundRequestMethods = new Map(); + // Every outbound request still awaiting a response, keyed by JSON-RPC id. + // Serves two readers: the `markResponseRejected` correlation (#1953), which + // needs the method, and the connection diagnostics (#2318), which need the + // method and when it went out. Entries are dropped as responses arrive; a + // request the SDK gave up on (a timeout, a cancel) stays until the session + // resets, because it is still unanswered and that is what the diagnostics + // report. + private outstandingRequests = new Map(); + // The id of the most recently answered request per method, for + // `markResponseRejected` (#1953). At most one id per method, so it does not + // grow with the session. private lastAnsweredRequestByMethod = new Map(); + // Connection diagnostics (#2318): the most recent response to one of our + // requests, and the most recent long-lived transport stream (the standalone + // `GET` on Streamable HTTP). Both per-session, reset with the maps above. + private lastResponse: LastResponse | undefined; + private notificationStream: TrackedNotificationStream | undefined; /** True when the cached transport was built with an OAuth authProvider attached. */ private transportHasAuthProvider = false; /** Dedupes concurrent ambient auth challenges (reason + scopes). */ @@ -1006,6 +1036,47 @@ export class InspectorClient extends InspectorClientEventTarget { this.clientInfo, Object.keys(clientOptions).length > 0 ? clientOptions : undefined, ); + this.annotateSdkRequestTimeouts(this.client); + } + + /** + * Decorate the SDK Client's `request` so a per-request timeout rejects with + * the connection's state at that moment instead of a bare `Request timed + * out` (#2318). + * + * `Protocol.request` is the one funnel every SDK verb goes through — + * `listTools`, `callTool`, `readResource`, the handshake's `initialize`, + * and this class's own raw `client.request` calls all reach the wire + * through it — so decorating it here covers the whole surface without a + * `catch` at each of the several dozen call sites, and covers the next + * verb added without anyone remembering to. The SDK offers no hook on the + * rejection path: the timeout error is constructed and thrown inside + * `request()` itself. + * + * Replaced on the instance rather than through a subclass because + * `request` is declared with two overloads (spec method with a keyed + * result, custom method with an explicit schema), and an `override` would + * have to restate both against types that move with every SDK release. One + * implementation signature serves both at runtime; the single cast on the + * way back is what the overloads cost. + */ + private annotateSdkRequestTimeouts(client: Client): void { + type SdkRequest = Client["request"]; + const sdkRequest: SdkRequest = client.request.bind(client); + const decorated = async ( + ...args: Parameters + ): Promise => { + try { + return await sdkRequest(...args); + } catch (err) { + throw annotateRequestTimeout( + err, + args[0].method, + this.getConnectionDiagnostics(), + ); + } + }; + client.request = decorated as SdkRequest; } /** @@ -1043,7 +1114,12 @@ export class InspectorClient extends InspectorClientEventTarget { return { trackRequest: (message: JSONRPCRequest, origin: MessageOrigin) => { if (origin === "client") { - this.outboundRequestMethods.set(message.id, message.method); + this.outstandingRequests.set(message.id, { + id: message.id, + method: message.method, + sentAt: Date.now(), + }); + this.dispatchConnectionDiagnosticsChange(); } const entry: MessageEntry = { id: crypto.randomUUID(), @@ -1065,11 +1141,13 @@ export class InspectorClient extends InspectorClientEventTarget { // frame answers no specific request, so it is skipped. const responseId = message.id; if (origin === "server" && responseId !== undefined) { - const method = this.outboundRequestMethods.get(responseId); - this.outboundRequestMethods.delete(responseId); + const method = this.outstandingRequests.get(responseId)?.method; + this.outstandingRequests.delete(responseId); if (method !== undefined) { this.lastAnsweredRequestByMethod.set(method, responseId); + this.lastResponse = { method, receivedAt: Date.now() }; } + this.dispatchConnectionDiagnosticsChange(); } const entry: MessageEntry = { id: crypto.randomUUID(), @@ -1928,8 +2006,11 @@ export class InspectorClient extends InspectorClientEventTarget { // start-clean path rather than in `disconnect()` for the reason documented // above: one route out (`onerror` with no `onclose`) tears down nothing // (#1953). - this.outboundRequestMethods.clear(); + this.outstandingRequests.clear(); this.lastAnsweredRequestByMethod.clear(); + this.lastResponse = undefined; + this.notificationStream = undefined; + this.dispatchConnectionDiagnosticsChange(); // Per-session for the same reason: both name entries of the PREVIOUS // server's list. Cleared here as well as in `disconnect()` because the // route out that tears down nothing (`onerror` with no `onclose`) would @@ -2098,10 +2179,14 @@ export class InspectorClient extends InspectorClientEventTarget { }, onFetchRequest: (entry: FetchRequestEntryBase) => { this.dispatchFetchRequest({ ...entry, category: "transport" }); + this.noteTransportStream(entry); }, onFetchResponseBody: (id: string, body: string) => { this.dispatchFetchRequestBodyUpdate(id, body); }, + onFetchStreamUpdate: (id: string, stream: FetchStreamState) => { + this.dispatchFetchRequestStreamUpdate(id, stream); + }, ...(this.serverSettings && { settings: this.serverSettings }), }; if (this.isHttpOAuthConfig() && oauthManager) { @@ -5962,6 +6047,80 @@ export class InspectorClient extends InspectorClientEventTarget { this.dispatchTypedEvent("fetchRequestBodyUpdate", { id, responseBody }); } + private dispatchFetchRequestStreamUpdate( + id: string, + stream: FetchStreamState, + ): void { + this.dispatchTypedEvent("fetchRequestStreamUpdate", { id, stream }); + // The diagnostics follow only the stream they are currently watching; + // an update for an older one (a reconnected stream's predecessor + // closing late) changes nothing they report. + if (this.notificationStream?.id !== id) return; + this.notificationStream = { + ...this.notificationStream, + eventCount: stream.eventCount, + ...(stream.closedAt && { closedAt: stream.closedAt.getTime() }), + }; + this.dispatchConnectionDiagnosticsChange(); + } + + /** + * Adopt a transport fetch as the notification stream when it is one — a + * `GET` answered with an unbounded event stream — so the diagnostics can + * say how long it has been open and what it has delivered (#2318). The + * newest such stream wins: the SDK reopens the standalone `GET` after a + * drop, and the current one is the one a stalled request is queued behind. + */ + private noteTransportStream(entry: FetchRequestEntryBase): void { + if ( + !isLongLivedStreamResponse( + entry.method, + findHeader(entry.responseHeaders, "content-type"), + ) + ) { + return; + } + this.notificationStream = { + id: entry.id, + url: entry.url, + openedAt: entry.timestamp.getTime(), + eventCount: 0, + }; + this.dispatchConnectionDiagnosticsChange(); + } + + /** + * What this client is still waiting on, when it last heard back, and the + * state of the notification stream (#2318). A fresh snapshot per call — + * the same one `annotateRequestTimeout` writes a timeout message from, and + * the one `connectionDiagnosticsChange` carries. + */ + getConnectionDiagnostics(): ConnectionDiagnostics { + return { + capturedAt: Date.now(), + outstandingRequests: [...this.outstandingRequests.values()], + ...(this.lastResponse && { lastResponse: { ...this.lastResponse } }), + ...(this.notificationStream && { + notificationStream: this.publicStreamState(this.notificationStream), + }), + }; + } + + /** The stream state minus the entry id, which is bookkeeping, not a fact. */ + private publicStreamState( + stream: TrackedNotificationStream, + ): NotificationStreamState { + const { id: _id, ...state } = stream; + return state; + } + + private dispatchConnectionDiagnosticsChange(): void { + this.dispatchTypedEvent( + "connectionDiagnosticsChange", + this.getConnectionDiagnostics(), + ); + } + /** * Get current session ID (from OAuth state authId) */ diff --git a/core/mcp/inspectorClientEventTarget.ts b/core/mcp/inspectorClientEventTarget.ts index 58aa41f6..f9ce6c8c 100644 --- a/core/mcp/inspectorClientEventTarget.ts +++ b/core/mcp/inspectorClientEventTarget.ts @@ -20,6 +20,7 @@ import type { MessageEntry, StderrLogEntry, FetchRequestEntry, + FetchStreamState, PromptGetInvocation, ResourceReadInvocation, ResourceTemplateReadInvocation, @@ -28,6 +29,7 @@ import type { RequestMetadata, } from "./types.js"; import type { MalformedListItem } from "./listSalvage.js"; +import type { ConnectionDiagnostics } from "./connectionDiagnostics.js"; import type { Tool, ServerCapabilities, @@ -92,6 +94,20 @@ export interface InspectorClientEventMap { fetchRequest: FetchRequestEntry; /** Fired when an in-flight fetch's response body is read asynchronously. */ fetchRequestBodyUpdate: { id: string; responseBody: string }; + /** + * Fired as a tracked long-lived stream (the standalone `GET` on Streamable + * HTTP, the legacy SSE event stream) delivers events and when it ends. The + * body of such a stream is never captured, so this is the log's only view + * of its lifetime (#2318). + */ + fetchRequestStreamUpdate: { id: string; stream: FetchStreamState }; + /** + * Fired whenever the connection diagnostics snapshot changes — a request + * goes out, a response comes back, the notification stream opens, delivers + * an event or closes, or a new session resets it. Carries the fresh + * snapshot; `getConnectionDiagnostics()` returns the same one (#2318). + */ + connectionDiagnosticsChange: ConnectionDiagnostics; /** * Fired whenever the client transitions `status` to `"error"` from a path * that is NOT an awaited promise — i.e. a mid-session transport failure diff --git a/core/mcp/inspectorClientProtocol.ts b/core/mcp/inspectorClientProtocol.ts index 53c7134e..8eefd758 100644 --- a/core/mcp/inspectorClientProtocol.ts +++ b/core/mcp/inspectorClientProtocol.ts @@ -40,6 +40,7 @@ import type { } from "@modelcontextprotocol/client"; import type { JsonValue } from "../json/jsonUtils.js"; import type { MalformedListItem } from "./listSalvage.js"; +import type { ConnectionDiagnostics } from "./connectionDiagnostics.js"; import type { InspectorClientEventTarget } from "./inspectorClientEventTarget.js"; import type { SkillEntry } from "./skillsSchemas.js"; import type { SkillsExtensionSupport } from "./skills.js"; @@ -78,6 +79,14 @@ export interface InspectorClientProtocol extends InspectorClientEventTarget { getDiscoverResult(): DiscoverResult | undefined; getServerSettings(): InspectorServerSettings | undefined; setServerSettings(settings: InspectorServerSettings): void; + /** + * What the client is still waiting on, when it last heard back, and the + * state of the notification stream — the snapshot a request timeout is + * annotated with (#2318). Optional so existing test doubles satisfy the + * interface without implementing it; `connectionDiagnosticsChange` fires + * whenever it changes. + */ + getConnectionDiagnostics?(): ConnectionDiagnostics; getAppRendererClient(): AppRendererClient | null; // Connection control diff --git a/core/mcp/node/transport.ts b/core/mcp/node/transport.ts index dd3c8358..e0dadcbf 100644 --- a/core/mcp/node/transport.ts +++ b/core/mcp/node/transport.ts @@ -49,6 +49,7 @@ export function createTransportNode( pipeStderr = false, onFetchRequest, onFetchResponseBody, + onFetchStreamUpdate, authProvider, settings, interceptAuthChallenges = false, @@ -118,6 +119,7 @@ export function createTransportNode( ? createFetchTracker(sseFetch, { trackRequest: onFetchRequest, updateResponseBody: onFetchResponseBody, + updateStream: onFetchStreamUpdate, }) : sseFetch; @@ -138,6 +140,7 @@ export function createTransportNode( ? createFetchTracker(fetchWithOptionalAuthIntercept, { trackRequest: onFetchRequest, updateResponseBody: onFetchResponseBody, + updateStream: onFetchStreamUpdate, }) : fetchWithOptionalAuthIntercept; @@ -165,6 +168,7 @@ export function createTransportNode( ? createFetchTracker(fetchWithOptionalAuthIntercept, { trackRequest: onFetchRequest, updateResponseBody: onFetchResponseBody, + updateStream: onFetchStreamUpdate, }) : fetchWithOptionalAuthIntercept; diff --git a/core/mcp/remote/createRemoteTransport.ts b/core/mcp/remote/createRemoteTransport.ts index 96803d8b..d93f9b06 100644 --- a/core/mcp/remote/createRemoteTransport.ts +++ b/core/mcp/remote/createRemoteTransport.ts @@ -58,6 +58,7 @@ export function createRemoteTransport( onStderr: transportOptions.onStderr, onFetchRequest: transportOptions.onFetchRequest, onFetchResponseBody: transportOptions.onFetchResponseBody, + onFetchStreamUpdate: transportOptions.onFetchStreamUpdate, authProvider: transportOptions.authProvider, settings: transportOptions.settings, }, diff --git a/core/mcp/remote/node/remote-session.ts b/core/mcp/remote/node/remote-session.ts index fc7c7654..2dba31f6 100644 --- a/core/mcp/remote/node/remote-session.ts +++ b/core/mcp/remote/node/remote-session.ts @@ -4,7 +4,7 @@ import type { Transport } from "@modelcontextprotocol/client"; import type { JSONRPCMessage } from "@modelcontextprotocol/client"; -import type { FetchRequestEntryBase } from "../../types.js"; +import type { FetchRequestEntryBase, FetchStreamState } from "../../types.js"; import type { RemoteEvent } from "../types.js"; import type { AuthChallenge } from "../../../auth/challenge.js"; import { AuthChallengeError } from "../../../auth/challenge.js"; @@ -288,6 +288,17 @@ export class RemoteSession { }); } + onFetchStreamUpdate(id: string, stream: FetchStreamState): void { + this.pushEvent({ + type: "fetch_stream_update", + data: { + id, + eventCount: stream.eventCount, + ...(stream.closedAt && { closedAt: stream.closedAt.toISOString() }), + }, + }); + } + onStderr(entry: { timestamp: Date; message: string }): void { this.pushEvent({ type: "stdio_log", diff --git a/core/mcp/remote/node/server.ts b/core/mcp/remote/node/server.ts index 15179eb1..f08a1f8b 100644 --- a/core/mcp/remote/node/server.ts +++ b/core/mcp/remote/node/server.ts @@ -868,6 +868,8 @@ export function createRemoteApp( onFetchRequest: (entry) => session.onFetchRequest(entry), onFetchResponseBody: (id, body) => session.onFetchResponseBody(id, body), + onFetchStreamUpdate: (id, stream) => + session.onFetchStreamUpdate(id, stream), authProvider: authHandle?.provider, settings: body.settings, // Always intercept 401/403 on the node MCP transport. Without this, the diff --git a/core/mcp/remote/remoteClientTransport.ts b/core/mcp/remote/remoteClientTransport.ts index 9cd22460..6a0ba873 100644 --- a/core/mcp/remote/remoteClientTransport.ts +++ b/core/mcp/remote/remoteClientTransport.ts @@ -12,7 +12,7 @@ import type { MessageExtraInfo, } from "@modelcontextprotocol/client"; import type { InspectorServerSettings, StderrLogEntry } from "../types.js"; -import type { FetchRequestEntryBase } from "../types.js"; +import type { FetchRequestEntryBase, FetchStreamState } from "../types.js"; import type { AuthChallenge, AuthChallengeOutcome, @@ -60,6 +60,9 @@ export interface RemoteTransportOptions { /** Callback for async response-body updates to a previously tracked fetch. */ onFetchResponseBody?: (id: string, responseBody: string) => void; + /** Callback for a previously tracked long-lived stream's events and close (#2318). */ + onFetchStreamUpdate?: (id: string, stream: FetchStreamState) => void; + /** Optional OAuth client provider for Bearer authentication */ authProvider?: import("@modelcontextprotocol/client").OAuthClientProvider; @@ -545,6 +548,15 @@ export class RemoteClientTransport implements Transport { parsed.data.id, parsed.data.responseBody, ); + } else if ( + parsed.type === "fetch_stream_update" && + this.options.onFetchStreamUpdate + ) { + const { id, eventCount, closedAt } = parsed.data; + this.options.onFetchStreamUpdate(id, { + eventCount, + ...(closedAt !== undefined && { closedAt: new Date(closedAt) }), + }); } else if (parsed.type === "stdio_log" && this.options.onStderr) { this.options.onStderr({ timestamp: new Date(parsed.data.timestamp), diff --git a/core/mcp/remote/types.ts b/core/mcp/remote/types.ts index 9e948401..661f1522 100644 --- a/core/mcp/remote/types.ts +++ b/core/mcp/remote/types.ts @@ -137,6 +137,7 @@ export type RemoteEventType = | "message" | "fetch_request" | "fetch_request_body_update" + | "fetch_stream_update" | "stdio_log" | "transport_error" | "auth_challenge"; @@ -156,6 +157,16 @@ export interface RemoteEventFetchRequestBodyUpdate { data: { id: string; responseBody: string }; } +/** + * A long-lived stream's lifecycle, forwarded from the backend's fetch tracker + * (#2318). `closedAt` is an ISO string on the wire; the browser transport + * rebuilds the `Date` the way it does for `fetch_request`'s timestamp. + */ +export interface RemoteEventFetchStreamUpdate { + type: "fetch_stream_update"; + data: { id: string; eventCount: number; closedAt?: string }; +} + export interface RemoteEventStdioLog { type: "stdio_log"; data: { timestamp: string; message: string }; @@ -178,6 +189,7 @@ export type RemoteEvent = | RemoteEventMessage | RemoteEventFetchRequest | RemoteEventFetchRequestBodyUpdate + | RemoteEventFetchStreamUpdate | RemoteEventStdioLog | RemoteEventTransportError | RemoteEventAuthChallenge; diff --git a/core/mcp/state/fetchRequestLogState.ts b/core/mcp/state/fetchRequestLogState.ts index 080fd39b..2e547ad6 100644 --- a/core/mcp/state/fetchRequestLogState.ts +++ b/core/mcp/state/fetchRequestLogState.ts @@ -151,6 +151,28 @@ export class FetchRequestLogState extends TypedEventTarget, + ): void => { + const { id, stream } = event.detail; + const idx = this.fetchRequests.findIndex((e) => e.id === id); + if (idx === -1) return; + this.fetchRequests[idx] = { ...this.fetchRequests[idx]!, stream }; + this.dispatchTypedEvent("fetchRequestsChange", this.getFetchRequests()); + }; + this.client.addEventListener( + "fetchRequestStreamUpdate", + onFetchRequestStreamUpdate, + ); + const sessionStorage = options.sessionStorage; const sessionId = options.sessionId; @@ -194,6 +216,10 @@ export class FetchRequestLogState extends TypedEventTarget; @@ -966,6 +988,14 @@ export interface CreateTransportOptions { */ onFetchResponseBody?: (id: string, responseBody: string) => void; + /** + * Optional callback fired asynchronously as a previously tracked long-lived + * stream (GET + `text/event-stream`) delivers events and when it ends. The + * tracker never buffers such a body, so this is the only view the consumer + * gets of the stream's lifetime (#2318). + */ + onFetchStreamUpdate?: (id: string, stream: FetchStreamState) => void; + /** * Optional OAuth client provider for Bearer authentication (SSE, streamable-http). * When set, the SDK injects tokens and handles 401 via the provider. diff --git a/core/react/useInspectorClient.ts b/core/react/useInspectorClient.ts index 1d22a94c..448144c5 100644 --- a/core/react/useInspectorClient.ts +++ b/core/react/useInspectorClient.ts @@ -12,6 +12,7 @@ import type { } from "@modelcontextprotocol/client"; import type { ExcludedTool } from "../mcp/types.js"; import type { MalformedListItem } from "../mcp/listSalvage.js"; +import type { ConnectionDiagnostics } from "../mcp/connectionDiagnostics.js"; import { useStoreSnapshot } from "./useStoreSnapshot.js"; // Module-scope frozen object so the `?? EMPTY_CLIENT_CAPABILITIES` @@ -27,6 +28,13 @@ const EMPTY_CLIENT_CAPABILITIES: ClientCapabilities = Object.freeze({}); */ const NO_EXCLUDED_TOOLS: ExcludedTool[] = []; const NO_MALFORMED_LIST_ITEMS: MalformedListItem[] = []; +// Also the fallback for a client that does not implement the optional +// accessor (a test double), so the field is always a snapshot and never +// `undefined` — one shape for consumers, whatever client is attached. +const NO_CONNECTION_DIAGNOSTICS: ConnectionDiagnostics = Object.freeze({ + capturedAt: 0, + outstandingRequests: [], +}); const readStatus = (client: InspectorClientProtocol): ConnectionStatus => client.getStatus(); @@ -53,6 +61,10 @@ const readExcludedTools = (client: InspectorClientProtocol): ExcludedTool[] => const readMalformedListItems = ( client: InspectorClientProtocol, ): MalformedListItem[] => client.getMalformedListItems(); +const readConnectionDiagnostics = ( + client: InspectorClientProtocol, +): ConnectionDiagnostics => + client.getConnectionDiagnostics?.() ?? NO_CONNECTION_DIAGNOSTICS; export interface UseInspectorClientResult { status: ConnectionStatus; @@ -86,6 +98,13 @@ export interface UseInspectorClientResult { * non-empty set means that list rendered without them (#1909). */ malformedListItems: MalformedListItem[]; + /** + * What the client is still waiting on, when it last heard back, and the + * state of the notification stream — re-read on every request, response + * and stream event. The Connection Info modal renders it; a request + * timeout's message is written from the same snapshot (#2318). + */ + connectionDiagnostics: ConnectionDiagnostics; /** * Message from the most recent mid-session transport failure (the client's * `error` event — stdio crash, SSE drop, HTTP 5xx). Stays set until the next @@ -174,6 +193,12 @@ export function useInspectorClient( readMalformedListItems, NO_MALFORMED_LIST_ITEMS, ); + const connectionDiagnostics = useStoreSnapshot( + inspectorClient, + "connectionDiagnosticsChange", + readConnectionDiagnostics, + NO_CONNECTION_DIAGNOSTICS, + ); // `lastError` is the one value here that is NOT a snapshot of something the // client stores: there is no `getLastError()`, because the client emits the @@ -243,6 +268,7 @@ export function useInspectorClient( discoverResult, excludedTools, malformedListItems, + connectionDiagnostics, lastError, appRendererClient: inspectorClient?.getAppRendererClient() ?? null, connect,