Merge remote-tracking branch 'origin/v2/main' into v2/fix/2295-pathless-skill-uri
This commit is contained in:
cliffhall
commit
bee2b3f463
17 changed files with 3357 additions and 62 deletions
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
252
core/mcp/connectionDiagnostics.ts
Normal file
252
core/mcp/connectionDiagnostics.ts
Normal file
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -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<string, string>,
|
||||
/**
|
||||
* 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<string, string> | 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;
|
||||
|
|
@ -215,11 +222,13 @@ export function isLongLivedStreamResponse(
|
|||
method: string,
|
||||
contentType: string | null | undefined,
|
||||
): boolean {
|
||||
if (method !== "GET") return false;
|
||||
if (method.toUpperCase() !== "GET") return false;
|
||||
if (!contentType) return false;
|
||||
// Media types are case-insensitive (RFC 9110 §8.3.1); a server may send
|
||||
// `Text/Event-Stream` and still be sending an event stream.
|
||||
const type = contentType.toLowerCase();
|
||||
return (
|
||||
contentType.includes("text/event-stream") ||
|
||||
contentType.includes("application/x-ndjson")
|
||||
type.includes("text/event-stream") || type.includes("application/x-ndjson")
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -233,6 +242,194 @@ 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;
|
||||
/**
|
||||
* How often, at most, `updateStream` reports a moving event count, in ms.
|
||||
* The first report (open, zero events) and the close are immediate; counts
|
||||
* in between are coalesced so a server pushing hundreds of notifications a
|
||||
* second costs one update per window rather than one per event — each
|
||||
* update re-publishes the whole fetch log and, on the web client, crosses
|
||||
* the backend-to-browser event stream. Defaults to
|
||||
* `DEFAULT_STREAM_UPDATE_INTERVAL_MS`; `0` reports every event, for tests.
|
||||
*/
|
||||
streamUpdateIntervalMs?: number;
|
||||
}
|
||||
|
||||
/** See `FetchTrackingCallbacks.streamUpdateIntervalMs`. */
|
||||
export const DEFAULT_STREAM_UPDATE_INTERVAL_MS = 250;
|
||||
|
||||
/**
|
||||
* The framing a long-lived stream's events arrive in — the two content types
|
||||
* `isLongLivedStreamResponse` accepts. SSE dispatches on a blank line;
|
||||
* NDJSON dispatches on every line.
|
||||
*/
|
||||
export type LongLivedStreamFraming = "sse" | "ndjson";
|
||||
|
||||
/** The framing for a long-lived stream's content type. */
|
||||
export function longLivedStreamFraming(
|
||||
contentType: string | null | undefined,
|
||||
): LongLivedStreamFraming {
|
||||
return contentType?.toLowerCase().includes("application/x-ndjson")
|
||||
? "ndjson"
|
||||
: "sse";
|
||||
}
|
||||
|
||||
/**
|
||||
* Count the events on a long-lived stream without buffering it, and report
|
||||
* when it ends.
|
||||
*
|
||||
* Reads a `clone()` of the response through a reader and discards the bytes,
|
||||
* so the transport keeps consuming the original at its own pace (the same
|
||||
* tee the bounded-body path relies on). Counting follows the framing the
|
||||
* transport's own parser applies. For SSE a blank line dispatches the block
|
||||
* before it, and only a block carrying a `data` field is an event —
|
||||
* keepalive comments (`: ping`) and bare `event:`/`id:` lines are not; a
|
||||
* partial block at end-of-stream is dropped, exactly as the SSE spec and the
|
||||
* SDK's `EventSourceParserStream` (which has no flush) drop it. For NDJSON
|
||||
* every non-blank line is one event.
|
||||
*
|
||||
* Reports once as soon as watching starts (zero events, open), once per event,
|
||||
* and once more on every exit from the read loop — a clean end-of-stream, a
|
||||
* 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,
|
||||
framing: LongLivedStreamFraming,
|
||||
callbacks: FetchTrackingCallbacks,
|
||||
): void {
|
||||
const listener = callbacks.updateStream;
|
||||
if (!listener) return;
|
||||
// Every report goes through here: the listener is consumer code (a relay
|
||||
// sink that may already be closed, a state store), and it runs from three
|
||||
// places none of which may throw outward — synchronously inside the fetch
|
||||
// wrapper (the open), from a timer (a coalesced count), and off the
|
||||
// discarded watcher promise (the close). A throw is the listener's
|
||||
// problem; the stream's bookkeeping carries on.
|
||||
const report = (state: FetchStreamState): void => {
|
||||
try {
|
||||
listener(id, state);
|
||||
} catch {
|
||||
// See above.
|
||||
}
|
||||
};
|
||||
const intervalMs =
|
||||
callbacks.streamUpdateIntervalMs ?? DEFAULT_STREAM_UPDATE_INTERVAL_MS;
|
||||
let body: ReadableStream<Uint8Array> | 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;
|
||||
// A moving count is reported at most once per interval (see
|
||||
// `streamUpdateIntervalMs`); the open and the close go out at once.
|
||||
let reportTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
const reportCount = (): void => {
|
||||
if (intervalMs <= 0) {
|
||||
report({ eventCount });
|
||||
return;
|
||||
}
|
||||
if (reportTimer !== null) return;
|
||||
reportTimer = setTimeout(() => {
|
||||
reportTimer = null;
|
||||
report({ eventCount });
|
||||
}, intervalMs);
|
||||
};
|
||||
// Announce the stream as open before a byte arrives. A stream that never
|
||||
// delivers anything is the #2187 tell, and it would otherwise be the one
|
||||
// stream with no state to show — the count only moves on an event.
|
||||
report({ eventCount });
|
||||
|
||||
const consumeLine = (line: string): void => {
|
||||
if (framing === "ndjson") {
|
||||
if (line.trim() === "") return;
|
||||
eventCount += 1;
|
||||
reportCount();
|
||||
return;
|
||||
}
|
||||
if (line === "") {
|
||||
if (blockHasData) {
|
||||
eventCount += 1;
|
||||
blockHasData = false;
|
||||
reportCount();
|
||||
}
|
||||
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;
|
||||
};
|
||||
|
||||
// Split off every complete line in `pending`. The SSE grammar (and the
|
||||
// SDK's parser) accept LF, CRLF and a bare CR as a line ending, so all three
|
||||
// end a line here; a CR that is the last byte so far is held back, since
|
||||
// the LF of a CRLF pair may still be in flight in the next chunk.
|
||||
const consumeCompleteLines = (): void => {
|
||||
for (;;) {
|
||||
const cr = pending.indexOf("\r");
|
||||
const lf = pending.indexOf("\n");
|
||||
const end = cr === -1 ? lf : lf === -1 ? cr : Math.min(cr, lf);
|
||||
if (end === -1) return;
|
||||
if (pending[end] === "\r") {
|
||||
if (end === pending.length - 1) return;
|
||||
const width = pending[end + 1] === "\n" ? 2 : 1;
|
||||
consumeLine(pending.slice(0, end));
|
||||
pending = pending.slice(end + width);
|
||||
} else {
|
||||
consumeLine(pending.slice(0, end));
|
||||
pending = pending.slice(end + 1);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const pump = async (): Promise<void> => {
|
||||
try {
|
||||
for (;;) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
pending += decoder.decode(value, { stream: true });
|
||||
consumeCompleteLines();
|
||||
}
|
||||
// A CR held back as a possible half of a CRLF is, at end-of-stream, a
|
||||
// whole line ending — `data: x\r\r` is a complete event. Anything after
|
||||
// it is a partial block and is dropped, as above.
|
||||
if (pending.endsWith("\r")) consumeLine(pending.slice(0, -1));
|
||||
} 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.
|
||||
}
|
||||
if (reportTimer !== null) {
|
||||
clearTimeout(reportTimer);
|
||||
reportTimer = null;
|
||||
}
|
||||
report({ eventCount, closedAt: new Date() });
|
||||
};
|
||||
// Deliberately not awaited: the fetch wrapper has to return the response
|
||||
// now, and this watcher lives as long as the stream does. It owns its own
|
||||
// failures — the read loop is wrapped in a `try`, and every report goes
|
||||
// through the non-throwing `report` above — so the promise cannot reject.
|
||||
void pump();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -257,7 +454,12 @@ export function createFetchTracker(
|
|||
: input instanceof URL
|
||||
? input.toString()
|
||||
: input.url;
|
||||
const method = init?.method || "GET";
|
||||
// The method decides the long-lived-stream classification below, so it
|
||||
// has to be the one the request actually carries: a `Request` input
|
||||
// carries its own, and HTTP methods compare case-insensitively.
|
||||
const method = (
|
||||
init?.method ?? (input instanceof Request ? input.method : "GET")
|
||||
).toUpperCase();
|
||||
|
||||
// Extract headers, redacting sensitive values BEFORE they reach any
|
||||
// downstream sink (logger, in-memory list, persisted session storage).
|
||||
|
|
@ -378,6 +580,18 @@ 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,
|
||||
longLivedStreamFraming(response.headers.get("content-type")),
|
||||
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
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ import type {
|
|||
MessageOrigin,
|
||||
FetchRequestEntry,
|
||||
FetchRequestEntryBase,
|
||||
FetchStreamState,
|
||||
InspectorServerSettings,
|
||||
ResourceReadInvocation,
|
||||
ResourceTemplateReadInvocation,
|
||||
|
|
@ -265,10 +266,31 @@ 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 { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
||||
import {
|
||||
annotateRequestTimeout,
|
||||
isRequestTimeoutError,
|
||||
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;
|
||||
|
|
@ -490,13 +512,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<string | number, string>();
|
||||
// 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<string | number, OutstandingRequest>();
|
||||
// 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<string, string | number>();
|
||||
// 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). */
|
||||
|
|
@ -1021,6 +1053,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<SdkRequest>
|
||||
): Promise<unknown> => {
|
||||
try {
|
||||
return await sdkRequest(...args);
|
||||
} catch (err) {
|
||||
throw annotateRequestTimeout(
|
||||
err,
|
||||
args[0].method,
|
||||
this.getConnectionDiagnostics(),
|
||||
);
|
||||
}
|
||||
};
|
||||
client.request = decorated as SdkRequest;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1058,7 +1131,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(),
|
||||
|
|
@ -1069,6 +1147,20 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
};
|
||||
this.dispatchTypedEvent("message", entry);
|
||||
},
|
||||
trackSendFailure: (message: JSONRPCRequest, error: unknown) => {
|
||||
// A send that failed because the request timed out is not a send
|
||||
// that failed: the browser's remote transport awaits the response
|
||||
// inside `send`, so its relay timeout surfaces here — the request
|
||||
// reached the server and went unanswered, which is the state to
|
||||
// keep reporting. Every other failure means the frame never reached
|
||||
// the wire, and leaving it would report a request the server never
|
||||
// saw as unanswered, in every timeout message and Connection Info
|
||||
// row, until the session resets (#2318).
|
||||
if (isRequestTimeoutError(error)) return;
|
||||
if (this.outstandingRequests.delete(message.id)) {
|
||||
this.dispatchConnectionDiagnosticsChange();
|
||||
}
|
||||
},
|
||||
trackResponse: (
|
||||
message: JSONRPCResultResponse | JSONRPCErrorResponse,
|
||||
origin: MessageOrigin,
|
||||
|
|
@ -1080,11 +1172,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(),
|
||||
|
|
@ -1943,8 +2037,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
|
||||
|
|
@ -2113,10 +2210,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) {
|
||||
|
|
@ -2857,8 +2958,19 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
const raw = await new Promise<unknown>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
this.pendingRawWireRequests.delete(id);
|
||||
// The same error, with the same annotation, as an SDK request that
|
||||
// times out: this path bypasses `Protocol.request`, so it builds the
|
||||
// SDK's own timeout shape and runs it through the decorator's
|
||||
// annotation by hand — a raw-wire caller sees one kind of timeout,
|
||||
// not two (#2318).
|
||||
reject(
|
||||
new Error(`Raw request "${method}" timed out after ${timeoutMs} ms`),
|
||||
annotateRequestTimeout(
|
||||
new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", {
|
||||
timeout: timeoutMs,
|
||||
}),
|
||||
method,
|
||||
this.getConnectionDiagnostics(),
|
||||
),
|
||||
);
|
||||
}, timeoutMs);
|
||||
this.pendingRawWireRequests.set(id, { resolve, reject, timer });
|
||||
|
|
@ -2868,7 +2980,17 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
clearTimeout(pending.timer);
|
||||
this.pendingRawWireRequests.delete(id);
|
||||
}
|
||||
reject(err instanceof Error ? err : new Error(String(err)));
|
||||
// The browser's remote transport awaits the response inside `send`,
|
||||
// so its relay wait can expire here first, as the SDK's timeout
|
||||
// shape; annotate it exactly as the local timer above does (a
|
||||
// non-timeout error passes through untouched).
|
||||
reject(
|
||||
annotateRequestTimeout(
|
||||
err instanceof Error ? err : new Error(String(err)),
|
||||
method,
|
||||
this.getConnectionDiagnostics(),
|
||||
),
|
||||
);
|
||||
});
|
||||
});
|
||||
return resultSchema.parse(raw);
|
||||
|
|
@ -5989,6 +6111,87 @@ 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,
|
||||
// The entry's `timestamp` is when the request went out; the stream is
|
||||
// open from when the response headers arrived, `duration` later.
|
||||
openedAt: entry.timestamp.getTime() + (entry.duration ?? 0),
|
||||
eventCount: 0,
|
||||
};
|
||||
this.dispatchConnectionDiagnosticsChange();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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(),
|
||||
// Copies, not the tracked records: the snapshot is handed to callers
|
||||
// and carried on errors, and a mutation there must not reach the
|
||||
// correlation state.
|
||||
outstandingRequests: [...this.outstandingRequests.values()].map(
|
||||
(request) => ({ ...request }),
|
||||
),
|
||||
...(this.lastResponse && { lastResponse: { ...this.lastResponse } }),
|
||||
...(this.notificationStream && {
|
||||
notificationStream: this.publicStreamState(this.notificationStream),
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
/** 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)
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -24,6 +24,19 @@ export interface MessageTrackingCallbacks {
|
|||
message: JSONRPCNotification,
|
||||
origin: MessageOrigin,
|
||||
) => void;
|
||||
/**
|
||||
* An outgoing request the base transport failed to send (the connection
|
||||
* closed, the fetch failed before the frame reached the wire). Fires after
|
||||
* `trackRequest` already recorded it, so a consumer keeping "requests still
|
||||
* awaiting a response" can roll that entry back — a request that never went
|
||||
* out is not unanswered (#2318). Not fired when the caller aborted the
|
||||
* request through `requestSignal` *during* the send: the modern transport
|
||||
* aborts an in-flight request's own stream to cancel it (and the SDK does
|
||||
* the same on a per-request timeout), and a request that was sent and then
|
||||
* given up on is exactly the unanswered kind. A signal already aborted
|
||||
* before the send began is a failed send like any other.
|
||||
*/
|
||||
trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -95,7 +108,19 @@ export class MessageTrackingTransport implements Transport {
|
|||
"client",
|
||||
);
|
||||
} else if ("method" in message) {
|
||||
this.callbacks.trackRequest?.(message as JSONRPCRequest, "client");
|
||||
const request = message as JSONRPCRequest;
|
||||
this.callbacks.trackRequest?.(request, "client");
|
||||
// A signal already aborted here means the frame will not go out at
|
||||
// all — that is a failed send. One that becomes aborted while the
|
||||
// send is in flight is a cancellation of a request that may well
|
||||
// have reached the server, which stays unanswered.
|
||||
const abortedBeforeSend = options?.requestSignal?.aborted === true;
|
||||
return this.baseTransport.send(message, options).catch((err) => {
|
||||
if (abortedBeforeSend || !options?.requestSignal?.aborted) {
|
||||
this.callbacks.trackSendFailure?.(request, err);
|
||||
}
|
||||
throw err;
|
||||
});
|
||||
}
|
||||
} else if ("method" in message) {
|
||||
this.callbacks.trackNotification?.(
|
||||
|
|
|
|||
|
|
@ -49,11 +49,18 @@ export function createTransportNode(
|
|||
pipeStderr = false,
|
||||
onFetchRequest,
|
||||
onFetchResponseBody,
|
||||
onFetchStreamUpdate,
|
||||
authProvider,
|
||||
settings,
|
||||
interceptAuthChallenges = false,
|
||||
onAuthChallengeObserved,
|
||||
} = options;
|
||||
// Any of the three tracking callbacks earns a tracker: each is optional on
|
||||
// the tracker itself, and a caller that wants only stream updates (or only
|
||||
// bodies) must not be silently handed the bare fetch (#2318).
|
||||
const wantsFetchTracking = Boolean(
|
||||
onFetchRequest || onFetchResponseBody || onFetchStreamUpdate,
|
||||
);
|
||||
|
||||
// `optionsFetchFn` is the caller's whole fetch stack and already sits on top of
|
||||
// a proxy-aware base when one is needed — the Node clients install
|
||||
|
|
@ -114,10 +121,11 @@ export function createTransportNode(
|
|||
const sseFetch = configuredSseFetch
|
||||
? withChallengeObserver(configuredSseFetch)
|
||||
: fetchWithOptionalAuthIntercept;
|
||||
const trackedFetch = onFetchRequest
|
||||
const trackedFetch = wantsFetchTracking
|
||||
? createFetchTracker(sseFetch, {
|
||||
trackRequest: onFetchRequest,
|
||||
updateResponseBody: onFetchResponseBody,
|
||||
updateStream: onFetchStreamUpdate,
|
||||
})
|
||||
: sseFetch;
|
||||
|
||||
|
|
@ -134,10 +142,11 @@ export function createTransportNode(
|
|||
...(headers && { headers }),
|
||||
};
|
||||
|
||||
const postFetch = onFetchRequest
|
||||
const postFetch = wantsFetchTracking
|
||||
? createFetchTracker(fetchWithOptionalAuthIntercept, {
|
||||
trackRequest: onFetchRequest,
|
||||
updateResponseBody: onFetchResponseBody,
|
||||
updateStream: onFetchStreamUpdate,
|
||||
})
|
||||
: fetchWithOptionalAuthIntercept;
|
||||
|
||||
|
|
@ -161,10 +170,11 @@ export function createTransportNode(
|
|||
...(headers && { headers }),
|
||||
};
|
||||
|
||||
const transportFetch = onFetchRequest
|
||||
const transportFetch = wantsFetchTracking
|
||||
? createFetchTracker(fetchWithOptionalAuthIntercept, {
|
||||
trackRequest: onFetchRequest,
|
||||
updateResponseBody: onFetchResponseBody,
|
||||
updateStream: onFetchStreamUpdate,
|
||||
})
|
||||
: fetchWithOptionalAuthIntercept;
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
@ -328,6 +328,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",
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -11,8 +11,10 @@ import type {
|
|||
JSONRPCMessage,
|
||||
MessageExtraInfo,
|
||||
} from "@modelcontextprotocol/client";
|
||||
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
||||
import { findHeader, isLongLivedStreamResponse } from "../fetchTracking.js";
|
||||
import type { InspectorServerSettings, StderrLogEntry } from "../types.js";
|
||||
import type { FetchRequestEntryBase } from "../types.js";
|
||||
import type { FetchRequestEntryBase, FetchStreamState } from "../types.js";
|
||||
import type {
|
||||
AuthChallenge,
|
||||
AuthChallengeOutcome,
|
||||
|
|
@ -61,6 +63,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;
|
||||
|
||||
|
|
@ -261,6 +266,14 @@ export class RemoteClientTransport implements Transport {
|
|||
private eventStreamConsumeTask: Promise<void> | null = null;
|
||||
private restartingEventStream = false;
|
||||
private closed = false;
|
||||
/**
|
||||
* Long-lived streams the backend has reported open, by fetch entry id, with
|
||||
* the last event count seen for each. `close()` reports them closed
|
||||
* itself: it aborts the event channel before asking the backend to
|
||||
* disconnect, so the backend watcher's own close report lands on an
|
||||
* unconsumed queue and is lost with the session (#2318).
|
||||
*/
|
||||
private openStreams = new Map<string, number>();
|
||||
private readonly sseResponseWaits = new Map<
|
||||
string | number,
|
||||
SseResponseWait
|
||||
|
|
@ -539,12 +552,23 @@ export class RemoteClientTransport implements Transport {
|
|||
this.resetSseResponseWait(progressToken);
|
||||
}
|
||||
this.onmessage?.(msg, undefined);
|
||||
} else if (
|
||||
parsed.type === "fetch_request" &&
|
||||
this.options.onFetchRequest
|
||||
) {
|
||||
} else if (parsed.type === "fetch_request") {
|
||||
const entry = parsed.data;
|
||||
this.options.onFetchRequest({
|
||||
// Stream bookkeeping runs whether or not the consumer logs
|
||||
// requests: `onFetchStreamUpdate` is documented to work on its
|
||||
// own, and `close()` needs to know which streams are open. It
|
||||
// exists only to serve that callback, though — without one the
|
||||
// set would only ever grow, one entry per reconnect.
|
||||
if (
|
||||
this.options.onFetchStreamUpdate &&
|
||||
isLongLivedStreamResponse(
|
||||
entry.method,
|
||||
findHeader(entry.responseHeaders, "content-type"),
|
||||
)
|
||||
) {
|
||||
this.openStreams.set(entry.id, 0);
|
||||
}
|
||||
this.options.onFetchRequest?.({
|
||||
...entry,
|
||||
timestamp:
|
||||
typeof entry.timestamp === "string"
|
||||
|
|
@ -559,6 +583,19 @@ 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;
|
||||
if (closedAt !== undefined) this.openStreams.delete(id);
|
||||
else if (this.openStreams.has(id)) {
|
||||
this.openStreams.set(id, eventCount);
|
||||
}
|
||||
this.options.onFetchStreamUpdate(id, {
|
||||
eventCount,
|
||||
...(closedAt !== undefined && { closedAt: new Date(closedAt) }),
|
||||
});
|
||||
} else if (parsed.type === "stdio_log" && this.options.onStderr) {
|
||||
this.options.onStderr({
|
||||
timestamp: new Date(parsed.data.timestamp),
|
||||
|
|
@ -693,10 +730,17 @@ export class RemoteClientTransport implements Transport {
|
|||
const arm = () => {
|
||||
timer = setTimeout(() => {
|
||||
this.sseResponseWaits.delete(requestId);
|
||||
// The SDK's own per-request timeout shape, not a plain `Error`:
|
||||
// this wait can expire before the SDK's timer does (its budget is
|
||||
// the relay's, not the request's), and the rejection travels up
|
||||
// through `Client.request`, where `InspectorClient` annotates a
|
||||
// timeout of exactly this shape with the connection's state
|
||||
// (#2318). A plain error would pass that decorator untouched and
|
||||
// reach the user as a bare timeout on the web client alone.
|
||||
reject(
|
||||
new Error(
|
||||
`Timed out waiting for MCP response on SSE (${this.sseResponseTimeoutMs}ms)`,
|
||||
),
|
||||
new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", {
|
||||
timeout: this.sseResponseTimeoutMs,
|
||||
}),
|
||||
);
|
||||
}, this.sseResponseTimeoutMs);
|
||||
};
|
||||
|
|
@ -836,6 +880,13 @@ export class RemoteClientTransport implements Transport {
|
|||
this.closed = true;
|
||||
this.cancelAllSseWaits(new Error("Transport closed"));
|
||||
await this.stopEventStream();
|
||||
// The disconnect below ends every stream the backend still holds open,
|
||||
// and its close reports cannot reach us any more (see `openStreams`).
|
||||
const closedAt = new Date();
|
||||
for (const [id, eventCount] of this.openStreams) {
|
||||
this.options.onFetchStreamUpdate?.(id, { eventCount, closedAt });
|
||||
}
|
||||
this.openStreams.clear();
|
||||
|
||||
if (this._sessionId) {
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -119,7 +119,10 @@ export class RemoteInspectorClientStorage implements InspectorClientStorage {
|
|||
|
||||
const data = (await res.json()) as InspectorClientSessionState;
|
||||
|
||||
// Deserialize state (convert ISO strings back to Date objects)
|
||||
// Deserialize state (convert ISO strings back to Date objects). The
|
||||
// nested `stream.closedAt` is a Date too, and `JSON.stringify` turned it
|
||||
// into an ISO string on the way out; left as a string it would fail the
|
||||
// `getTime()` every consumer of the type performs (#2318).
|
||||
return {
|
||||
...data,
|
||||
fetchRequests: data.fetchRequests.map((req) => ({
|
||||
|
|
@ -130,6 +133,9 @@ export class RemoteInspectorClientStorage implements InspectorClientStorage {
|
|||
: req.timestamp instanceof Date
|
||||
? req.timestamp
|
||||
: new Date(req.timestamp),
|
||||
...(req.stream?.closedAt !== undefined && {
|
||||
stream: { ...req.stream, closedAt: new Date(req.stream.closedAt) },
|
||||
}),
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -151,6 +151,28 @@ export class FetchRequestLogState extends TypedEventTarget<FetchRequestLogStateE
|
|||
onFetchRequestBodyUpdate,
|
||||
);
|
||||
|
||||
// A long-lived stream's lifecycle arrives the same way its sibling's
|
||||
// body does: asynchronously, patched onto the entry in place (#2318). A
|
||||
// missing entry is left silent here, unlike the body path — a stream
|
||||
// reports on every event, so after a clear the stragglers are many and
|
||||
// none of them is data the user could have wanted retained.
|
||||
const onFetchRequestStreamUpdate = (
|
||||
event: TypedEventGeneric<
|
||||
InspectorClientEventMap,
|
||||
"fetchRequestStreamUpdate"
|
||||
>,
|
||||
): 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<FetchRequestLogStateE
|
|||
"fetchRequestBodyUpdate",
|
||||
onFetchRequestBodyUpdate,
|
||||
);
|
||||
this.client.removeEventListener(
|
||||
"fetchRequestStreamUpdate",
|
||||
onFetchRequestStreamUpdate,
|
||||
);
|
||||
this.client.removeEventListener("saveSession", onSaveSession);
|
||||
}
|
||||
this.client = null;
|
||||
|
|
@ -220,6 +246,10 @@ export class FetchRequestLogState extends TypedEventTarget<FetchRequestLogStateE
|
|||
"fetchRequestBodyUpdate",
|
||||
onFetchRequestBodyUpdate,
|
||||
);
|
||||
this.client.removeEventListener(
|
||||
"fetchRequestStreamUpdate",
|
||||
onFetchRequestStreamUpdate,
|
||||
);
|
||||
}
|
||||
this.client = null;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -380,10 +380,32 @@ export interface FetchRequestEntry {
|
|||
responseBody?: string;
|
||||
duration?: number; // Time between request and response in ms
|
||||
error?: string;
|
||||
/**
|
||||
* Lifecycle of a long-lived stream response (the standalone `GET` on
|
||||
* Streamable HTTP, the legacy SSE event stream), whose body is never
|
||||
* captured. Filled in asynchronously via `fetchRequestStreamUpdate` as
|
||||
* events arrive and when the stream ends; absent on every bounded response
|
||||
* (#2318).
|
||||
*/
|
||||
stream?: FetchStreamState;
|
||||
/** Distinguishes OAuth/auth fetches from MCP transport fetches */
|
||||
category: FetchRequestCategory;
|
||||
}
|
||||
|
||||
/**
|
||||
* What the fetch tracker knows about a long-lived stream it declined to
|
||||
* buffer: how many SSE events have been delivered on it so far, and when it
|
||||
* ended. `closedAt` is absent while the stream is still open. The Network
|
||||
* tab renders it in place of the body, and `InspectorClient` folds the most
|
||||
* recent transport stream into its connection diagnostics (#2318).
|
||||
*/
|
||||
export interface FetchStreamState {
|
||||
/** SSE events delivered so far — dispatched events, not keepalive comments. */
|
||||
eventCount: number;
|
||||
/** When the stream ended (server close, network error, or abort), once it has. */
|
||||
closedAt?: Date;
|
||||
}
|
||||
|
||||
/** Entry shape from createFetchTracker before category is added by the caller */
|
||||
export type FetchRequestEntryBase = Omit<FetchRequestEntry, "category">;
|
||||
|
||||
|
|
@ -986,6 +1008,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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Reference in a new issue