fix: address Copilot review round 1 on #2318

- fetchTracking: count NDJSON streams line by line instead of leaving
  them at zero forever; justify the detached watcher promise; document
  that a partial SSE block at end-of-stream is dropped, as the SDK's own
  parser (no flush) and the SSE spec drop it.
- uncapturedBodyNote: case-insensitive content-type lookup, and a
  stream label (SSE / NDJSON) for the Network badge.
- InspectorClient: `openedAt` is when the response headers arrived, not
  when the GET went out; the snapshot copies its request records; a
  request the transport failed to send is rolled back out of the
  unanswered set via a new `trackSendFailure` callback; a raw-wire
  timeout is built as the SDK's own timeout shape and annotated the
  same way, so callers see one kind of timeout.
- Remote session storage revives the nested `stream.closedAt` Date.
- Hook test models a legacy double with `Object.defineProperty`, not a
  double cast.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01APKSYHGxXBUpnR7ErFiLJn
Signed-off-by: cliffhall <cliff@futurescale.com>
This commit is contained in:
cliffhall 2026-09-13 22:03:00 -04:00
commit 1a5dd6820d
13 changed files with 267 additions and 20 deletions

View file

@ -17,6 +17,7 @@ import { RiErrorWarningLine } from "react-icons/ri";
import type { FetchRequestEntry } from "@inspector/core/mcp/types.js";
import {
isLongLivedStreamEntry,
longLivedStreamLabel,
uncapturedBodyNote,
} from "../../../utils/uncapturedBodyNote";
import { ContentViewer } from "../../elements/ContentViewer/ContentViewer";
@ -549,7 +550,9 @@ export function NetworkEntry({
{entry.duration != null && (
<DurationText>{formatDuration(entry.duration)}</DurationText>
)}
{isLongLivedStreamEntry(entry) && <StreamBadge>SSE</StreamBadge>}
{isLongLivedStreamEntry(entry) && (
<StreamBadge>{longLivedStreamLabel(entry)}</StreamBadge>
)}
<Badge color={statusColor(entry)} variant="status">
{statusLabel(entry)}
</Badge>

View file

@ -792,6 +792,40 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => {
expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]);
});
it("counts every non-blank line on an NDJSON stream", async () => {
const server = controlledStream();
const updates: Array<{ id: string; stream: FetchStreamState }> = [];
const fetcher = createFetchTracker(
(async () =>
new Response(server.stream, {
headers: { "content-type": "application/x-ndjson" },
})) as typeof fetch,
{ updateStream: (id, state) => updates.push({ id, stream: state }) },
);
await fetcher("https://example.com/mcp", { method: "GET" });
server.push('{"jsonrpc":"2.0","method":"a"}\n\n');
server.push('{"jsonrpc":"2.0","method":"b"}\r\n');
// A partial line is not an event until its newline arrives.
server.push('{"jsonrpc":"2.0",');
await flush();
expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]);
server.push('"method":"c"}\n');
await flush();
expect(updates.at(-1)!.stream.eventCount).toBe(3);
});
it("drops a partial SSE block at end-of-stream, as the SDK's parser does", async () => {
const server = controlledStream();
const { updates } = await trackStream(server.stream);
// A data line with no dispatching blank line before the stream ends.
server.push("data: one\n\ndata: two\n");
server.end();
await flush();
const last = updates.at(-1)!.stream;
expect(last.eventCount).toBe(1);
expect(last.closedAt).toBeInstanceOf(Date);
});
it("reports the close when the server ends the stream", async () => {
const server = controlledStream();
const { updates, id } = await trackStream(server.stream);

View file

@ -1,4 +1,5 @@
import { describe, it, expect, vi } from "vitest";
import { SdkErrorCode } from "@modelcontextprotocol/client";
import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js";
import { ModernGetTaskResultSchema } from "@inspector/core/mcp/modernTaskSchemas.js";
@ -134,7 +135,15 @@ describe("InspectorClient raw-wire channel (#1631)", () => {
{},
ModernGetTaskResultSchema,
);
const assertion = expect(promise).rejects.toThrow(/timed out/);
// The same annotated shape an SDK request's timeout carries (#2318):
// the SDK's code, the budget, the method, and the connection state.
const assertion = expect(promise).rejects.toMatchObject({
code: SdkErrorCode.RequestTimeout,
message: expect.stringMatching(
/^Request timed out after 10ms \(tasks\/get\)\. No requests are unanswered\./,
),
data: expect.objectContaining({ timeout: 10, method: "tasks/get" }),
});
await vi.advanceTimersByTimeAsync(20);
await assertion;
} finally {

View file

@ -33,6 +33,30 @@ function makeTracked() {
}
describe("MessageTrackingTransport.send", () => {
it("reports a request the base transport failed to send, after tracking it (#2318)", async () => {
const { callbacks, base, tracked } = makeTracked();
const trackSendFailure = vi.fn();
const failing = new MessageTrackingTransport(base, {
...callbacks,
trackSendFailure,
});
const boom = new Error("connection closed");
base.send = async () => {
throw boom;
};
const request = { jsonrpc: "2.0", id: 1, method: "tools/list" } as const;
await expect(failing.send(request)).rejects.toBe(boom);
expect(callbacks.trackRequest).toHaveBeenCalledWith(request, "client");
expect(trackSendFailure).toHaveBeenCalledWith(request, boom);
// Only requests roll back: a failed notification has no pending entry.
trackSendFailure.mockClear();
await expect(
failing.send({ jsonrpc: "2.0", method: "notifications/initialized" }),
).rejects.toBe(boom);
expect(trackSendFailure).not.toHaveBeenCalled();
void tracked;
});
it("tracks an outgoing request", async () => {
const { callbacks, tracked } = makeTracked();
const request = { jsonrpc: "2.0", id: 1, method: "tools/list" } as const;

View file

@ -110,6 +110,45 @@ describe("RemoteInspectorClientStorage", () => {
);
});
it("loadSession revives a stream's closedAt alongside the entry timestamp (#2318)", async () => {
const fetchFn = vi.fn<typeof fetch>(
async () =>
new Response(
JSON.stringify({
fetchRequests: [
{
id: "open",
timestamp: "2026-01-01T00:00:00.000Z",
method: "GET",
url: "http://example.com/mcp",
requestHeaders: {},
stream: { eventCount: 2 },
},
{
id: "closed",
timestamp: "2026-01-01T00:00:00.000Z",
method: "GET",
url: "http://example.com/mcp",
requestHeaders: {},
stream: { eventCount: 3, closedAt: "2026-01-01T00:00:05.000Z" },
},
],
createdAt: 1735689600000,
updatedAt: 1735689600001,
}),
{ status: 200 },
),
);
const storage = makeStorage(fetchFn);
const state = await storage.loadSession("sid");
// An open stream's state passes through untouched.
expect(state?.fetchRequests[0]?.stream).toEqual({ eventCount: 2 });
const closed = state?.fetchRequests[1]?.stream;
expect(closed?.eventCount).toBe(3);
expect(closed?.closedAt).toBeInstanceOf(Date);
expect(closed?.closedAt?.toISOString()).toBe("2026-01-01T00:00:05.000Z");
});
it("loadSession preserves Date timestamps already provided as Date instances", async () => {
const fetchFn = vi.fn<typeof fetch>(
async () =>

View file

@ -90,9 +90,9 @@ describe("useInspectorClient", () => {
const client = new FakeInspectorClient();
// `getConnectionDiagnostics` is optional on the protocol; a double that
// predates it must still yield the one shape consumers expect.
(
client as unknown as { getConnectionDiagnostics?: unknown }
).getConnectionDiagnostics = undefined;
Object.defineProperty(client, "getConnectionDiagnostics", {
value: undefined,
});
const { result } = renderHook(() => useInspectorClient(client));
expect(result.current.connectionDiagnostics).toEqual({
capturedAt: 0,

View file

@ -271,6 +271,21 @@ describe("InspectorClient request-timeout diagnostics (#2318)", () => {
const stream = c.getConnectionDiagnostics().notificationStream!;
expect(stream.url).toBe(server.url);
expect(stream.closedAt).toBeUndefined();
// Open from when the headers arrived, not from when the GET went out.
const getEntryAtOpen = log
.getFetchRequests()
.find((entry) => entry.method === "GET")!;
expect(stream.openedAt).toBe(
getEntryAtOpen.timestamp.getTime() + (getEntryAtOpen.duration ?? 0),
);
// The snapshot is a copy: mutating it does not reach the client.
const snapshot = c.getConnectionDiagnostics();
snapshot.outstandingRequests.push({
id: "x",
method: "bogus",
sentAt: 0,
});
expect(c.getConnectionDiagnostics().outstandingRequests).toEqual([]);
// A server push on the notification stream is an event.
server.pushEvent({

View file

@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
import type { FetchRequestEntry } from "@inspector/core/mcp/types.js";
import {
isLongLivedStreamEntry,
longLivedStreamLabel,
uncapturedBodyNote,
} from "./uncapturedBodyNote";
@ -26,6 +27,27 @@ describe("isLongLivedStreamEntry", () => {
isLongLivedStreamEntry({ ...streamEntry, responseHeaders: undefined }),
).toBe(false);
});
it("reads the content type whatever casing the entry recorded it with", () => {
expect(
isLongLivedStreamEntry({
...streamEntry,
responseHeaders: { "Content-Type": "text/event-stream" },
}),
).toBe(true);
});
});
describe("longLivedStreamLabel", () => {
it("names the framing the stream uses", () => {
expect(longLivedStreamLabel(streamEntry)).toBe("SSE");
expect(
longLivedStreamLabel({
...streamEntry,
responseHeaders: { "Content-Type": "application/x-ndjson" },
}),
).toBe("NDJSON");
});
});
describe("uncapturedBodyNote", () => {

View file

@ -1,17 +1,38 @@
// The Network entry's body placeholder for a response whose body was not
// captured (#2318). Pure: a string from the entry alone.
import { isLongLivedStreamResponse } from "@inspector/core/mcp/fetchTracking.js";
import {
findHeader,
isLongLivedStreamResponse,
longLivedStreamFraming,
} from "@inspector/core/mcp/fetchTracking.js";
import type { FetchRequestEntry } from "@inspector/core/mcp/types.js";
/** Whether the entry is an unbounded server-to-client stream (never buffered). */
/**
* Whether the entry is an unbounded server-to-client stream (never
* buffered). The header lookup is case-insensitive: a recorded entry keeps
* the wire casing (`Content-Type` on some hosts, and in a restored session),
* and a miss here would silently reclassify the stream as a bounded body.
*/
export function isLongLivedStreamEntry(entry: FetchRequestEntry): boolean {
return isLongLivedStreamResponse(
entry.method,
entry.responseHeaders?.["content-type"],
findHeader(entry.responseHeaders, "content-type"),
);
}
/**
* The badge label for a long-lived stream entry — `SSE` or `NDJSON`, by the
* same content-type split the tracker counts events with.
*/
export function longLivedStreamLabel(entry: FetchRequestEntry): string {
return longLivedStreamFraming(
findHeader(entry.responseHeaders, "content-type"),
) === "ndjson"
? "NDJSON"
: "SSE";
}
/**
* What stands in for a body that was not captured. A long-lived stream is
* never buffered, so its placeholder reports what the tracker learned by

View file

@ -251,15 +251,32 @@ export interface FetchTrackingCallbacks {
}
/**
* Count the SSE events on a long-lived stream without buffering it, and
* report when it ends.
* The framing a long-lived stream's events arrive in — the two content types
* `isLongLivedStreamResponse` accepts. SSE dispatches on a blank line;
* NDJSON dispatches on every line.
*/
export type LongLivedStreamFraming = "sse" | "ndjson";
/** The framing for a long-lived stream's content type. */
export function longLivedStreamFraming(
contentType: string | null | undefined,
): LongLivedStreamFraming {
return contentType?.includes("application/x-ndjson") ? "ndjson" : "sse";
}
/**
* Count the events on a long-lived stream without buffering it, and report
* when it ends.
*
* Reads a `clone()` of the response through a reader and discards the bytes,
* so the transport keeps consuming the original at its own pace (the same
* tee the bounded-body path relies on). Counting follows the SSE framing the
* transport's own parser applies: a blank line dispatches the block before
* it, and only a block carrying a `data` field is an event — keepalive
* comments (`: ping`) and bare `event:`/`id:` lines are not.
* tee the bounded-body path relies on). Counting follows the framing the
* transport's own parser applies. For SSE a blank line dispatches the block
* before it, and only a block carrying a `data` field is an event —
* keepalive comments (`: ping`) and bare `event:`/`id:` lines are not; a
* partial block at end-of-stream is dropped, exactly as the SSE spec and the
* SDK's `EventSourceParserStream` (which has no flush) drop it. For NDJSON
* every non-blank line is one event.
*
* Reports once as soon as watching starts (zero events, open), once per event,
* and once more on every exit from the read loop — a clean end-of-stream, a
@ -273,6 +290,7 @@ export interface FetchTrackingCallbacks {
function watchLongLivedStream(
response: Response,
id: string,
framing: LongLivedStreamFraming,
callbacks: FetchTrackingCallbacks,
): void {
const update = callbacks.updateStream;
@ -298,6 +316,12 @@ function watchLongLivedStream(
const consumeLine = (rawLine: string): void => {
const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
if (framing === "ndjson") {
if (line.trim() === "") return;
eventCount += 1;
update(id, { eventCount });
return;
}
if (line === "") {
if (blockHasData) {
eventCount += 1;
@ -331,6 +355,10 @@ function watchLongLivedStream(
}
update(id, { eventCount, closedAt: new Date() });
};
// Deliberately not awaited: the fetch wrapper has to return the response
// now, and this watcher lives as long as the stream does. It owns its own
// failures — the read loop is wrapped in a `try` that treats an error as a
// close — so the promise cannot reject.
void pump();
}
@ -481,7 +509,12 @@ export function createFetchTracker(
// but its event count and its eventual close are exactly the state a
// stalled connection is diagnosed from (#2318).
if (isLongLivedStream && response.body && !response.bodyUsed) {
watchLongLivedStream(response, id, callbacks);
watchLongLivedStream(
response,
id,
longLivedStreamFraming(response.headers.get("content-type")),
callbacks,
);
}
// Kick off a fire-and-forget read of the cloned body. The clone is an

View file

@ -270,6 +270,7 @@ import {
findHeader,
isLongLivedStreamResponse,
} from "./fetchTracking.js";
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
import {
annotateRequestTimeout,
type ConnectionDiagnostics,
@ -1130,6 +1131,15 @@ export class InspectorClient extends InspectorClientEventTarget {
};
this.dispatchTypedEvent("message", entry);
},
trackSendFailure: (message: JSONRPCRequest) => {
// The frame never reached the wire, so it is not awaiting a response.
// Leaving it would report a request the server never saw as
// unanswered, in every timeout message and Connection Info row, until
// the session resets (#2318).
if (this.outstandingRequests.delete(message.id)) {
this.dispatchConnectionDiagnosticsChange();
}
},
trackResponse: (
message: JSONRPCResultResponse | JSONRPCErrorResponse,
origin: MessageOrigin,
@ -2915,8 +2925,19 @@ export class InspectorClient extends InspectorClientEventTarget {
const raw = await new Promise<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 });
@ -6083,7 +6104,9 @@ export class InspectorClient extends InspectorClientEventTarget {
this.notificationStream = {
id: entry.id,
url: entry.url,
openedAt: entry.timestamp.getTime(),
// The entry's `timestamp` is when the request went out; the stream is
// open from when the response headers arrived, `duration` later.
openedAt: entry.timestamp.getTime() + (entry.duration ?? 0),
eventCount: 0,
};
this.dispatchConnectionDiagnosticsChange();
@ -6098,7 +6121,12 @@ export class InspectorClient extends InspectorClientEventTarget {
getConnectionDiagnostics(): ConnectionDiagnostics {
return {
capturedAt: Date.now(),
outstandingRequests: [...this.outstandingRequests.values()],
// Copies, not the tracked records: the snapshot is handed to callers
// and carried on errors, and a mutation there must not reach the
// correlation state.
outstandingRequests: [...this.outstandingRequests.values()].map(
(request) => ({ ...request }),
),
...(this.lastResponse && { lastResponse: { ...this.lastResponse } }),
...(this.notificationStream && {
notificationStream: this.publicStreamState(this.notificationStream),

View file

@ -24,6 +24,14 @@ export interface MessageTrackingCallbacks {
message: JSONRPCNotification,
origin: MessageOrigin,
) => void;
/**
* An outgoing request the base transport failed to send (the connection
* closed, the fetch failed before the frame reached the wire). Fires after
* `trackRequest` already recorded it, so a consumer keeping "requests still
* awaiting a response" can roll that entry back — a request that never went
* out is not unanswered (#2318).
*/
trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void;
}
/**
@ -95,7 +103,12 @@ export class MessageTrackingTransport implements Transport {
"client",
);
} else if ("method" in message) {
this.callbacks.trackRequest?.(message as JSONRPCRequest, "client");
const request = message as JSONRPCRequest;
this.callbacks.trackRequest?.(request, "client");
return this.baseTransport.send(message, options).catch((err) => {
this.callbacks.trackSendFailure?.(request, err);
throw err;
});
}
} else if ("method" in message) {
this.callbacks.trackNotification?.(

View file

@ -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) },
}),
})),
};
}