From d57801d7b3960e4f7cbed248387df39af998064b Mon Sep 17 00:00:00 2001 From: cliffhall Date: Sun, 13 Sep 2026 21:25:31 -0400 Subject: [PATCH] feat(core): report connection state on a request timeout (#2318) A request timeout used to surface as a bare `Request timed out`. Every fact needed to explain it was already in-process: what the client was still waiting on, when it last heard back, and the state of the standalone GET notification stream. The SDK's timeout now rejects with all three, and the same snapshot is shown live in Connection Info and on the Network entry for the stream. - core/mcp/connectionDiagnostics.ts: the snapshot type, its one-line description, and `annotateRequestTimeout`, which rebuilds the SDK's per-request timeout with the state at that moment. - InspectorClient decorates the SDK Client's `request` (the one funnel every verb goes through) so the annotation covers the whole surface; tracks outstanding requests with their send time, the last response, and the most recent long-lived transport stream; exposes `getConnectionDiagnostics()` and `connectionDiagnosticsChange`. - fetchTracking watches a long-lived stream through a tee'd clone, counting SSE events and reporting the close, via a new `updateStream` callback wired through the node transport, the remote session (`fetch_stream_update`), and the browser transport. - FetchRequestLogState patches `entry.stream`; the Network entry's placeholder reports the event count and open/closed state. - Connection Info gains a Connection Activity section, ticking while open (`useTickingClock`). Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01APKSYHGxXBUpnR7ErFiLJn Signed-off-by: cliffhall --- clients/web/src/App.test.tsx | 1 + clients/web/src/App.tsx | 2 + .../ConnectionInfoContent.stories.tsx | 45 +++ .../ConnectionInfoContent.test.tsx | 85 +++++ .../ConnectionInfoContent.tsx | 59 +++ .../ConnectionInfoModal.test.tsx | 21 ++ .../ConnectionInfoModal.tsx | 5 + .../NetworkEntry/NetworkEntry.stories.tsx | 37 ++ .../groups/NetworkEntry/NetworkEntry.test.tsx | 18 + .../groups/NetworkEntry/NetworkEntry.tsx | 20 +- clients/web/src/hooks/useTickingClock.test.ts | 58 +++ clients/web/src/hooks/useTickingClock.ts | 27 ++ .../core/mcp/connectionDiagnostics.test.ts | 279 ++++++++++++++ .../src/test/core/mcp/fetchTracking.test.ts | 199 +++++++++- .../mcp/state/fetchRequestLogState.test.ts | 45 +++ .../core/react/useInspectorClient.test.tsx | 36 ++ ...nspectorClient-timeout-diagnostics.test.ts | 351 ++++++++++++++++++ .../mcp/remote/remote-session.test.ts | 24 ++ .../remote/remoteClientTransport-unit.test.ts | 53 +++ .../web/src/utils/connectionActivity.test.ts | 113 ++++++ clients/web/src/utils/connectionActivity.ts | 63 ++++ .../web/src/utils/uncapturedBodyNote.test.ts | 57 +++ clients/web/src/utils/uncapturedBodyNote.ts | 29 ++ core/mcp/__tests__/fakeInspectorClient.ts | 16 + core/mcp/connectionDiagnostics.ts | 252 +++++++++++++ core/mcp/fetchTracking.ts | 114 +++++- core/mcp/inspectorClient.ts | 181 ++++++++- core/mcp/inspectorClientEventTarget.ts | 16 + core/mcp/inspectorClientProtocol.ts | 9 + core/mcp/node/transport.ts | 4 + core/mcp/remote/createRemoteTransport.ts | 1 + core/mcp/remote/node/remote-session.ts | 13 +- core/mcp/remote/node/server.ts | 2 + core/mcp/remote/remoteClientTransport.ts | 14 +- core/mcp/remote/types.ts | 12 + core/mcp/state/fetchRequestLogState.ts | 30 ++ core/mcp/types.ts | 30 ++ core/react/useInspectorClient.ts | 26 ++ 38 files changed, 2315 insertions(+), 32 deletions(-) create mode 100644 clients/web/src/hooks/useTickingClock.test.ts create mode 100644 clients/web/src/hooks/useTickingClock.ts create mode 100644 clients/web/src/test/core/mcp/connectionDiagnostics.test.ts create mode 100644 clients/web/src/test/integration/mcp/inspectorClient-timeout-diagnostics.test.ts create mode 100644 clients/web/src/utils/connectionActivity.test.ts create mode 100644 clients/web/src/utils/connectionActivity.ts create mode 100644 clients/web/src/utils/uncapturedBodyNote.test.ts create mode 100644 clients/web/src/utils/uncapturedBodyNote.ts create mode 100644 core/mcp/connectionDiagnostics.ts 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,