Merge pull request #1967 from modelcontextprotocol/v2/fix/modern-listen-stream-opens-on-listchanged
fix: open the modern listen stream whenever the filter is non-empty (#1920)
This commit is contained in:
Cliff Hall
GitHub
commit
75f4597281
4 changed files with 487 additions and 53 deletions
|
|
@ -52,7 +52,9 @@ export interface ResourceControlsProps {
|
|||
* Modern-era `subscriptions/listen` stream state (#1630). When `active`
|
||||
* (modern era with at least one subscription) the Subscriptions section shows
|
||||
* a stream-status badge in its panel and a status dot in its header. Legacy
|
||||
* connections pass `active: false` (or omit it) and see neither.
|
||||
* connections pass `active: false` (or omit it) and see neither — and so does
|
||||
* a stream open purely for list-change notifications, which this section has
|
||||
* nothing to say about (#1920).
|
||||
*/
|
||||
subscriptionStreamState?: ResourceSubscriptionStreamState;
|
||||
/** Negotiated protocol era; gates the modern subscription stream chrome. */
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import {
|
|||
type TestServerHttp,
|
||||
createTestServerInfo,
|
||||
createNumberedResources,
|
||||
createEchoTool,
|
||||
} from "@modelcontextprotocol/inspector-test-server";
|
||||
import type { ServerConfig } from "@modelcontextprotocol/inspector-test-server";
|
||||
import type { MessageEntry } from "@inspector/core/mcp/types.js";
|
||||
|
|
@ -23,6 +24,13 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
let server: TestServerHttp | null = null;
|
||||
|
||||
const RESOURCE_URI = "test://resource_0";
|
||||
/**
|
||||
* Every list-change opt-in off. The listen filter is config ∩ capability and
|
||||
* the SDK server advertises `listChanged` for every list it registers, so this
|
||||
* is what leaves the filter empty when nothing is subscribed — i.e. the
|
||||
* pre-#1920 world these lifecycle tests were written against.
|
||||
*/
|
||||
const NO_LIST_CHANGED = { tools: false, resources: false, prompts: false };
|
||||
const RESOURCE_URI_2 = "test://resource_1";
|
||||
|
||||
afterEach(async () => {
|
||||
|
|
@ -59,15 +67,28 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
return started;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param listChangedNotifications the Inspector-side opt-ins. The SDK server
|
||||
* advertises `listChanged` for every list it registers, so this — not the
|
||||
* server config — is how a test gets an *empty* listen filter: the filter is
|
||||
* config ∩ capability, and with every opt-in off only a subscribed URI can
|
||||
* open the stream (#1920).
|
||||
*/
|
||||
async function connect(
|
||||
url: string,
|
||||
era: "legacy" | "modern",
|
||||
listChangedNotifications?: {
|
||||
tools?: boolean;
|
||||
resources?: boolean;
|
||||
prompts?: boolean;
|
||||
},
|
||||
): Promise<{ connected: InspectorClient; messages: MessageEntry[] }> {
|
||||
const connected = new InspectorClient(
|
||||
{ type: "streamable-http", url },
|
||||
{
|
||||
environment: { transport: createTransportNode },
|
||||
versionNegotiation: eraToVersionNegotiation(era),
|
||||
...(listChangedNotifications ? { listChangedNotifications } : {}),
|
||||
},
|
||||
);
|
||||
const messages: MessageEntry[] = [];
|
||||
|
|
@ -103,6 +124,34 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
return params.notifications as Record<string, unknown> | undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* The private members these tests drive directly. `InspectorClient` declares
|
||||
* every one of them, so the shape below is a faithful mirror rather than a
|
||||
* reinterpretation — the double cast is only to reach past `private`, which no
|
||||
* single `as` can do (the two types share no public overlap). It is confined to
|
||||
* `internals()` so there is one such cast in the file, and it is unavoidable:
|
||||
* these are lifecycle branches no public API can reach against a healthy server
|
||||
* (there is no way to ask for a *remote* stream close, or for a `listen()` that
|
||||
* fails).
|
||||
*/
|
||||
interface StreamInternals {
|
||||
client: { listen: (...args: unknown[]) => Promise<McpSubscription> };
|
||||
modernSubscription: McpSubscription | null;
|
||||
modernListenGeneration: number;
|
||||
modernReconnectAttempts: number;
|
||||
subscribedResources: Set<string>;
|
||||
refreshModernSubscription(fromReconnect?: boolean): Promise<void>;
|
||||
onModernSubscriptionClosed(
|
||||
subscription: McpSubscription,
|
||||
reason: "local" | "graceful" | "remote",
|
||||
generation: number,
|
||||
): void;
|
||||
}
|
||||
|
||||
function internals(c: InspectorClient): StreamInternals {
|
||||
return c as unknown as StreamInternals;
|
||||
}
|
||||
|
||||
describe("modern era", () => {
|
||||
it("opens an acknowledged listen stream on subscribe (no resources/subscribe)", async () => {
|
||||
const started = await startServer({});
|
||||
|
|
@ -125,15 +174,43 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
});
|
||||
|
||||
it("closes the stream when the last subscription is removed", async () => {
|
||||
// Nothing else in the filter (no list-change opt-ins), so the last URI
|
||||
// leaving empties it and the stream closes (#1920).
|
||||
const started = await startServer({});
|
||||
const { connected } = await connect(started.url, "modern");
|
||||
const { connected, messages } = await connect(
|
||||
started.url,
|
||||
"modern",
|
||||
NO_LIST_CHANGED,
|
||||
);
|
||||
await connected.subscribeToResource(RESOURCE_URI);
|
||||
expect(connected.getResourceSubscriptionStreamState().active).toBe(true);
|
||||
|
||||
messages.length = 0;
|
||||
await connected.unsubscribeFromResource(RESOURCE_URI);
|
||||
const streamState = connected.getResourceSubscriptionStreamState();
|
||||
expect(streamState.active).toBe(false);
|
||||
expect(connected.getSubscribedResources()).toEqual([]);
|
||||
// No re-listen: there is nothing left to listen for.
|
||||
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
||||
});
|
||||
|
||||
it("keeps the stream open past the last subscription when a listChanged opt-in remains", async () => {
|
||||
// The other half of the above: with an advertised listChanged the filter
|
||||
// is still non-empty, so the last unsubscribe re-lists rather than
|
||||
// closing — and the state goes inactive because the *Subscriptions*
|
||||
// section has nothing to report, not because the stream is gone (#1920).
|
||||
const started = await startServer({});
|
||||
const { connected, messages } = await connect(started.url, "modern");
|
||||
await connected.subscribeToResource(RESOURCE_URI);
|
||||
|
||||
messages.length = 0;
|
||||
await connected.unsubscribeFromResource(RESOURCE_URI);
|
||||
|
||||
expect(connected.getSubscribedResources()).toEqual([]);
|
||||
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
||||
const filter = lastListenFilter(messages);
|
||||
expect(filter?.resourcesListChanged).toBe(true);
|
||||
expect(filter?.resourceSubscriptions).toBeUndefined();
|
||||
});
|
||||
|
||||
it("re-lists (stream stays open) when one of several URIs is removed", async () => {
|
||||
|
|
@ -197,6 +274,244 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
});
|
||||
});
|
||||
|
||||
// The stream used to be reachable only by subscribing to a resource, so a
|
||||
// server with no resources could never open it — and its `tools.listChanged`
|
||||
// notifications, which ride the same stream, could never arrive (#1920).
|
||||
describe("modern era: opening the stream for listChanged alone (#1920)", () => {
|
||||
async function startToolsOnlyServer(
|
||||
listChanged: ServerConfig["listChanged"],
|
||||
): Promise<TestServerHttp> {
|
||||
const started = createTestServerHttp({
|
||||
serverInfo: createTestServerInfo("tools-only-test", "1.0.0"),
|
||||
tools: [createEchoTool()],
|
||||
listChanged,
|
||||
modern: {},
|
||||
});
|
||||
await started.start();
|
||||
server = started;
|
||||
return started;
|
||||
}
|
||||
|
||||
it("opens the stream on connect against a tools-only server", async () => {
|
||||
const started = await startToolsOnlyServer({ tools: true });
|
||||
const { connected, messages } = await connect(started.url, "modern");
|
||||
|
||||
expect(connected.getCapabilities()?.resources).toBeUndefined();
|
||||
expect(methodsSent(messages)).toContain("subscriptions/listen");
|
||||
const filter = lastListenFilter(messages);
|
||||
expect(filter?.toolsListChanged).toBe(true);
|
||||
// Nothing is subscribed, so the filter carries no URIs at all.
|
||||
expect(filter?.resourceSubscriptions).toBeUndefined();
|
||||
// …and the Subscriptions section still has nothing to report.
|
||||
expect(connected.getSubscribedResources()).toEqual([]);
|
||||
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
||||
});
|
||||
|
||||
it("leaves the stream closed when the opt-in is disabled in config", async () => {
|
||||
// The filter is config ∩ capability, so turning the handler off in the
|
||||
// Inspector's own options is enough to keep the stream closed.
|
||||
const started = await startToolsOnlyServer({ tools: true });
|
||||
const connected = new InspectorClient(
|
||||
{ type: "streamable-http", url: started.url },
|
||||
{
|
||||
environment: { transport: createTransportNode },
|
||||
versionNegotiation: eraToVersionNegotiation("modern"),
|
||||
listChangedNotifications: { tools: false },
|
||||
},
|
||||
);
|
||||
const messages: MessageEntry[] = [];
|
||||
connected.addEventListener("message", (event) => {
|
||||
messages.push(event.detail);
|
||||
});
|
||||
await connected.connect();
|
||||
client = connected;
|
||||
|
||||
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
||||
});
|
||||
|
||||
it("opens no stream on the legacy era", async () => {
|
||||
// Legacy has no `subscriptions/listen` at all — list-change notifications
|
||||
// arrive on the session's own channel.
|
||||
const started = createTestServerHttp({
|
||||
serverInfo: createTestServerInfo("tools-only-legacy-test", "1.0.0"),
|
||||
tools: [createEchoTool()],
|
||||
listChanged: { tools: true },
|
||||
});
|
||||
await started.start();
|
||||
server = started;
|
||||
const { connected, messages } = await connect(started.url, "legacy");
|
||||
|
||||
expect(connected.getProtocolEra()).toBe("legacy");
|
||||
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
||||
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
||||
});
|
||||
|
||||
it("reconnects a dropped listChanged-only stream (no subscriptions to keep it alive)", async () => {
|
||||
const started = await startToolsOnlyServer({ tools: true });
|
||||
const { connected } = await connect(started.url, "modern");
|
||||
|
||||
const int = internals(connected);
|
||||
const dropped = int.modernSubscription;
|
||||
expect(dropped).not.toBeNull();
|
||||
if (!dropped) return;
|
||||
|
||||
int.onModernSubscriptionClosed(
|
||||
dropped,
|
||||
"remote",
|
||||
int.modernListenGeneration,
|
||||
);
|
||||
// Reconnect-by-re-listen runs on the filter, not on the subscribed set —
|
||||
// which is empty here, and used to be the reason to give up.
|
||||
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
||||
"reconnecting",
|
||||
);
|
||||
await vi.waitFor(() => {
|
||||
expect(int.modernSubscription).not.toBeNull();
|
||||
});
|
||||
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
||||
"acknowledged",
|
||||
);
|
||||
});
|
||||
|
||||
it("has the stream up before the connect event fires", async () => {
|
||||
// The managed list states start their initial `refresh()` from the
|
||||
// `connect` event, so the stream has to be established (or its retry
|
||||
// armed) first: dispatching earlier would let `tools/list` go out ahead of
|
||||
// `subscriptions/listen`, and a list the server changed in that window
|
||||
// would notify nobody.
|
||||
const started = await startToolsOnlyServer({ tools: true });
|
||||
const connected = new InspectorClient(
|
||||
{ type: "streamable-http", url: started.url },
|
||||
{
|
||||
environment: { transport: createTransportNode },
|
||||
versionNegotiation: eraToVersionNegotiation("modern"),
|
||||
},
|
||||
);
|
||||
let streamAtConnectEvent: boolean | undefined;
|
||||
connected.addEventListener("connect", () => {
|
||||
streamAtConnectEvent = internals(connected).modernSubscription !== null;
|
||||
});
|
||||
await connected.connect();
|
||||
client = connected;
|
||||
|
||||
expect(streamAtConnectEvent).toBe(true);
|
||||
});
|
||||
|
||||
it("does not announce the connection when a disconnect overtakes the stream open", async () => {
|
||||
// The listen round-trip widened the window between the handshake and the
|
||||
// `connect` announcement, so a `disconnect()` can land inside it.
|
||||
// Announcing anyway would restart every managed list refresh against a
|
||||
// session being torn down.
|
||||
const started = await startToolsOnlyServer({ tools: true });
|
||||
const connected = new InspectorClient(
|
||||
{ type: "streamable-http", url: started.url },
|
||||
{
|
||||
environment: { transport: createTransportNode },
|
||||
versionNegotiation: eraToVersionNegotiation("modern"),
|
||||
},
|
||||
);
|
||||
client = connected;
|
||||
let connectAnnounced = false;
|
||||
connected.addEventListener("connect", () => {
|
||||
connectAnnounced = true;
|
||||
});
|
||||
|
||||
// Hold the stream open so the disconnect lands while it is in flight.
|
||||
// `openStarted` is what makes the race deterministic: the stub is only
|
||||
// reached after the handshake, so disconnecting before it runs would leave
|
||||
// `releaseOpen` unset and hang the connect.
|
||||
const int = internals(connected);
|
||||
let releaseOpen: () => void = () => {};
|
||||
let markOpenStarted: () => void = () => {};
|
||||
const openStarted = new Promise<void>((resolve) => {
|
||||
markOpenStarted = resolve;
|
||||
});
|
||||
int.refreshModernSubscription = () => {
|
||||
markOpenStarted();
|
||||
return new Promise<void>((resolve) => {
|
||||
releaseOpen = resolve;
|
||||
});
|
||||
};
|
||||
|
||||
const connecting = connected.connect();
|
||||
await openStarted;
|
||||
const disconnecting = connected.disconnect();
|
||||
releaseOpen();
|
||||
await connecting;
|
||||
await disconnecting;
|
||||
|
||||
expect(connectAnnounced).toBe(false);
|
||||
});
|
||||
|
||||
it("connects anyway when the connect-time listen fails, and retries", async () => {
|
||||
// The handshake succeeded; every request-scoped feature works without the
|
||||
// stream, so the failure is handed to the reconnect machinery instead of
|
||||
// failing `connect()`.
|
||||
const started = await startToolsOnlyServer({ tools: true });
|
||||
const connected = new InspectorClient(
|
||||
{ type: "streamable-http", url: started.url },
|
||||
{
|
||||
environment: { transport: createTransportNode },
|
||||
versionNegotiation: eraToVersionNegotiation("modern"),
|
||||
},
|
||||
);
|
||||
// Shadow the private refresh on the instance so the connect-time open
|
||||
// fails without a live client to reach into (there is none until
|
||||
// `connect()` builds one). The bump mirrors the real method's contract —
|
||||
// it claims the generation synchronously before it can fail — which is
|
||||
// what the caller's guard reads to tell "our refresh failed" from "a newer
|
||||
// one took over".
|
||||
const int = internals(connected);
|
||||
int.refreshModernSubscription = () => {
|
||||
int.modernListenGeneration++;
|
||||
return Promise.reject(new Error("listen boom"));
|
||||
};
|
||||
|
||||
await expect(connected.connect()).resolves.toBeUndefined();
|
||||
client = connected;
|
||||
|
||||
expect(connected.getStatus()).toBe("connected");
|
||||
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
||||
"reconnecting",
|
||||
);
|
||||
});
|
||||
|
||||
it("leaves a superseded connect-time failure to the refresh that owns the stream", async () => {
|
||||
// Nothing about the ordering above makes this call the only refresh that
|
||||
// can be in flight: `statusChange` has already fired, and a concurrent
|
||||
// `subscribeToResource` — or a `disconnect()`, whose
|
||||
// `resetSubscriptionStream` bumps the generation too — can supersede this
|
||||
// one while its `listen()` is pending. Reconciling anyway would arm a
|
||||
// reconnect against a stream that is healthy (or a session that is gone),
|
||||
// so this makes the same ownership test the subscribe/unsubscribe paths
|
||||
// make. Driven here by a stub that bumps twice, standing in for "someone
|
||||
// else advanced it".
|
||||
const started = await startToolsOnlyServer({ tools: true });
|
||||
const connected = new InspectorClient(
|
||||
{ type: "streamable-http", url: started.url },
|
||||
{
|
||||
environment: { transport: createTransportNode },
|
||||
versionNegotiation: eraToVersionNegotiation("modern"),
|
||||
},
|
||||
);
|
||||
const int = internals(connected);
|
||||
// Two bumps: this call's own, plus the newer refresh that superseded it
|
||||
// before its failure landed.
|
||||
int.refreshModernSubscription = () => {
|
||||
int.modernListenGeneration += 2;
|
||||
return Promise.reject(new Error("listen boom"));
|
||||
};
|
||||
|
||||
await expect(connected.connect()).resolves.toBeUndefined();
|
||||
client = connected;
|
||||
|
||||
// No reconnect armed: the state is the newer refresh's to write.
|
||||
expect(connected.getResourceSubscriptionStreamState().status).not.toBe(
|
||||
"reconnecting",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("legacy era", () => {
|
||||
it("subscribes via resources/subscribe with no listen stream", async () => {
|
||||
const started = await startServer(undefined);
|
||||
|
|
@ -245,23 +560,6 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
// the client's private state — the pattern used across the InspectorClient
|
||||
// coverage-backfill suite — to drive them deterministically.
|
||||
describe("modern stream internals", () => {
|
||||
interface StreamInternals {
|
||||
client: { listen: (...args: unknown[]) => Promise<McpSubscription> };
|
||||
modernSubscription: McpSubscription | null;
|
||||
modernListenGeneration: number;
|
||||
modernReconnectAttempts: number;
|
||||
subscribedResources: Set<string>;
|
||||
onModernSubscriptionClosed(
|
||||
subscription: McpSubscription,
|
||||
reason: "local" | "graceful" | "remote",
|
||||
generation: number,
|
||||
): void;
|
||||
}
|
||||
|
||||
function internals(c: InspectorClient): StreamInternals {
|
||||
return c as unknown as StreamInternals;
|
||||
}
|
||||
|
||||
/** A controllable fake `McpSubscription` whose `closed` we resolve on demand. */
|
||||
function makeFakeSub(): {
|
||||
sub: McpSubscription;
|
||||
|
|
@ -355,8 +653,17 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
// Reachable without any close() failure: a rejecting `listen()` does it,
|
||||
// which is what a user subscribing while the reconnect timer re-lists
|
||||
// (i.e. exactly when the server is flaky) can produce.
|
||||
//
|
||||
// No list-change opt-ins, so connect leaves the stream closed (#1920) and
|
||||
// the first subscribe's `listen()` is reached synchronously — with a
|
||||
// stream to tear down first, the swap below would land after this call
|
||||
// already read `client.listen`.
|
||||
const started = await startServer({});
|
||||
const { connected } = await connect(started.url, "modern");
|
||||
const { connected } = await connect(
|
||||
started.url,
|
||||
"modern",
|
||||
NO_LIST_CHANGED,
|
||||
);
|
||||
const int = internals(connected);
|
||||
|
||||
const real = int.client.listen;
|
||||
|
|
@ -504,8 +811,15 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
});
|
||||
|
||||
it("reflects the subscription optimistically as 'connecting' before the ack", async () => {
|
||||
// No list-change opt-ins → no connect-time stream, so the subscribe's
|
||||
// `listen()` is the first one and this test's held ack is the one it waits
|
||||
// on (#1920).
|
||||
const started = await startServer({});
|
||||
const { connected } = await connect(started.url, "modern");
|
||||
const { connected } = await connect(
|
||||
started.url,
|
||||
"modern",
|
||||
NO_LIST_CHANGED,
|
||||
);
|
||||
const int = internals(connected);
|
||||
|
||||
// Hold the listen ack so we can observe the pre-ack (optimistic) state.
|
||||
|
|
@ -632,8 +946,14 @@ describe("resource subscriptions era fork (#1630)", () => {
|
|||
});
|
||||
|
||||
it("does not reconnect when the subscription set empties before the timer fires", async () => {
|
||||
// No list-change opt-ins, so emptying the set empties the *filter* —
|
||||
// which is what the timer's bail now tests (#1920).
|
||||
const started = await startServer({});
|
||||
const { connected } = await connect(started.url, "modern");
|
||||
const { connected } = await connect(
|
||||
started.url,
|
||||
"modern",
|
||||
NO_LIST_CHANGED,
|
||||
);
|
||||
await connected.subscribeToResource(RESOURCE_URI);
|
||||
const int = internals(connected);
|
||||
const fake = await installFakeSubscription(int);
|
||||
|
|
|
|||
|
|
@ -479,7 +479,10 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
// `McpSubscription` whose filter's `resourceSubscriptions` mirrors
|
||||
// `subscribedResources`; mutating the set re-lists (close old, open new), and
|
||||
// an unexpected `"remote"` close re-lists (reconnect-by-re-listen — the stream
|
||||
// is not resumable). `null` when no URI is subscribed or on the legacy era.
|
||||
// is not resumable). `null` on the legacy era, and on the modern era whenever
|
||||
// the filter is empty — which is *not* the same as "no URI subscribed": the
|
||||
// stream also carries the list-change opt-ins, so a tools-only server opens it
|
||||
// with no subscriptions at all (#1920).
|
||||
private modernSubscription: McpSubscription | null = null;
|
||||
// Monotonic guard so a stale re-list/reconnect (whose `listen()` or `closed`
|
||||
// resolves after a newer refresh already started) can detect it lost the race
|
||||
|
|
@ -1933,8 +1936,6 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
// capabilities and wipe tools/prompts/resources to empty on every connect.
|
||||
await this.fetchServerInfo();
|
||||
|
||||
this.dispatchTypedEvent("connect");
|
||||
|
||||
// Set initial logging level if configured and server supports it
|
||||
if (this.initialLoggingLevel && this.capabilities?.logging) {
|
||||
await this.client.setLoggingLevel(
|
||||
|
|
@ -2056,6 +2057,33 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
// a progress notification handler so the Protocol's _onprogress stays; timeout reset
|
||||
// and routing work, and we inject the caller's progressToken into dispatched events.
|
||||
}
|
||||
|
||||
// Modern era: the handlers registered above only fire for notifications
|
||||
// that reach us, and on this era every server→client notification rides
|
||||
// the `subscriptions/listen` stream. Open it now when the filter says
|
||||
// there is something to listen for — otherwise a list-change opt-in on a
|
||||
// server with no resources would have no way in, since a subscribe click
|
||||
// was the only thing that ever opened the stream (#1920).
|
||||
await this.openModernListenStreamOnConnect();
|
||||
|
||||
// Last, so the notification channel is established (or its retry armed)
|
||||
// before any consumer acts on the connection. The managed list states
|
||||
// start their initial `refresh()` from this event, so dispatching earlier
|
||||
// would let `tools/list` go out ahead of `subscriptions/listen` — and a
|
||||
// list the server changes in that window would notify nobody, leaving the
|
||||
// UI stale with no way to notice (#1920). The cost is one listen
|
||||
// round-trip added to connect on the modern era.
|
||||
//
|
||||
// …which is also why the announcement is conditional. Every await above
|
||||
// is a window for a `disconnect()` or a transport `onclose`/`onerror` to
|
||||
// overtake this connect, and the listen round-trip widened it. Announcing
|
||||
// then would restart every managed list refresh against a session being
|
||||
// torn down or already dead. `disconnecting` covers the teardown that has
|
||||
// claimed ownership but is still awaiting `client.close()` — the status is
|
||||
// whatever it was until that block finishes.
|
||||
if (this.status === "connected" && !this.disconnecting) {
|
||||
this.dispatchTypedEvent("connect");
|
||||
}
|
||||
} catch (error) {
|
||||
if (!isConnectAuthRecoveryError(error)) {
|
||||
this.status = "error";
|
||||
|
|
@ -4893,9 +4921,13 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
* opted-in notification type (SEP §7.4).
|
||||
*/
|
||||
private buildSubscriptionFilter(): SubscriptionFilter {
|
||||
const filter: SubscriptionFilter = {
|
||||
resourceSubscriptions: Array.from(this.subscribedResources),
|
||||
};
|
||||
const filter: SubscriptionFilter = {};
|
||||
// Omitted rather than sent empty: a listChanged-only stream (#1920) is not
|
||||
// subscribing to any resource, and `[]` would say it asked for none of a set
|
||||
// it is participating in.
|
||||
if (this.subscribedResources.size > 0) {
|
||||
filter.resourceSubscriptions = Array.from(this.subscribedResources);
|
||||
}
|
||||
if (
|
||||
this.listChangedNotifications.tools &&
|
||||
this.capabilities?.tools?.listChanged
|
||||
|
|
@ -4917,6 +4949,82 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
return filter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the modern listen stream should be open: the built filter carries
|
||||
* something to listen for (#1920). Before #1920 this was "at least one URI is
|
||||
* subscribed", which made the stream unreachable on a server with no resources
|
||||
* — a tools-only server advertising `tools.listChanged` had no way to open it,
|
||||
* so `notifications/tools/list_changed` could never arrive. The filter already
|
||||
* modelled the list-change opt-ins; only the trigger was narrower than the
|
||||
* filter it built. This matches the SDK's own `ClientOptions.listChanged`
|
||||
* auto-open, which opens whenever the effective (config ∩ capability)
|
||||
* intersection is non-empty.
|
||||
*
|
||||
* Note this is deliberately *not* the same predicate as the stream state's
|
||||
* `active` — see `modernStreamActive()`.
|
||||
*/
|
||||
private wantsModernStream(): boolean {
|
||||
const filter = this.buildSubscriptionFilter();
|
||||
return (
|
||||
(filter.resourceSubscriptions?.length ?? 0) > 0 ||
|
||||
filter.toolsListChanged === true ||
|
||||
filter.resourcesListChanged === true ||
|
||||
filter.promptsListChanged === true
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the stream state reports `active` — i.e. whether the *Subscriptions*
|
||||
* UI has a stream to describe. That is a narrower question than
|
||||
* `wantsModernStream()`: `ResourceSubscriptionStreamState` drives the
|
||||
* Subscriptions section's badge, so a stream open purely for list-change
|
||||
* notifications (no subscribed URI) has nothing to report there and stays
|
||||
* `active: false` (#1920). Keeping the two apart also preserves the invariant
|
||||
* the rest of this file is written against — an empty subscribed set is never
|
||||
* announced alongside an `active` stream.
|
||||
*/
|
||||
private modernStreamActive(): boolean {
|
||||
return this.subscribedResources.size > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Open the modern listen stream at the end of a successful connect, when the
|
||||
* filter is non-empty (#1920). Only the list-change opt-ins can make it
|
||||
* non-empty here — the subscribed set is emptied by `resetSessionState()` on
|
||||
* the way in — so this is exactly the "server advertises a listChanged the
|
||||
* Inspector wants" case that had no trigger before.
|
||||
*
|
||||
* A failure is not allowed to fail the connect: the handshake succeeded and
|
||||
* every request-scoped feature works without this stream. It is reported the
|
||||
* way a lost stream is — hand it to the reconnect machinery, which retries
|
||||
* with backoff and settles on `"ended"` past the cap.
|
||||
*
|
||||
* Gated on the same generation test as `subscribeToResource` — see the long
|
||||
* comment there. The `connect` event has deliberately *not* been dispatched
|
||||
* yet (that is the point of running here), so a list-state consumer is not the
|
||||
* risk; what is, is anything else that bumps the generation while this
|
||||
* `listen()` is in flight. `statusChange` has already fired, and any
|
||||
* concurrent call on this instance qualifies: a `subscribeToResource` from a
|
||||
* caller restoring subscriptions, or a `disconnect()` — whose
|
||||
* `resetSubscriptionStream` bumps the generation too, making a reconcile here
|
||||
* arm a reconnect for a session that is already gone.
|
||||
*/
|
||||
private async openModernListenStreamOnConnect(): Promise<void> {
|
||||
if (!this.isModernEra() || !this.wantsModernStream()) return;
|
||||
const generationBefore = this.modernListenGeneration;
|
||||
try {
|
||||
await this.refreshModernSubscription();
|
||||
} catch (error) {
|
||||
this.logger.error(
|
||||
{ error },
|
||||
"Failed to open the modern subscriptions/listen stream on connect",
|
||||
);
|
||||
if (this.modernListenGeneration === generationBefore + 1) {
|
||||
this.reconcileModernStreamStateAfterFailedRefresh();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Cancel a pending reconnect re-listen, if any (#1630). */
|
||||
private clearModernReconnectTimer(): void {
|
||||
if (this.modernReconnectTimer !== undefined) {
|
||||
|
|
@ -4927,9 +5035,10 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
|
||||
/**
|
||||
* (Re-)establish the modern `subscriptions/listen` stream to match the current
|
||||
* `subscribedResources` set (#1630). Because the stream is not resumable,
|
||||
* every filter change re-lists: the existing stream is closed and a fresh
|
||||
* `listen()` opened. With no subscribed URIs the stream is left closed.
|
||||
* filter (#1630). Because the stream is not resumable, every filter change
|
||||
* re-lists: the existing stream is closed and a fresh `listen()` opened. With
|
||||
* an empty filter — no subscribed URIs *and* no enabled list-change opt-in the
|
||||
* server advertises — the stream is left closed (#1920).
|
||||
*
|
||||
* `modernListenGeneration` guards against races — if a newer refresh starts
|
||||
* while this one awaits its acknowledgement, the just-opened stream is
|
||||
|
|
@ -4955,8 +5064,8 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
await closeSubscriptionBestEffort(previous);
|
||||
}
|
||||
|
||||
// Nothing subscribed → keep the stream closed.
|
||||
if (this.subscribedResources.size === 0) {
|
||||
// Nothing to listen for → keep the stream closed.
|
||||
if (!this.wantsModernStream()) {
|
||||
this.setModernStreamState(INACTIVE_SUBSCRIPTION_STREAM_STATE);
|
||||
return;
|
||||
}
|
||||
|
|
@ -4978,7 +5087,7 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
// starts fresh next time (#1630).
|
||||
this.modernReconnectAttempts = 0;
|
||||
this.setModernStreamState({
|
||||
active: true,
|
||||
active: this.modernStreamActive(),
|
||||
status: "acknowledged",
|
||||
honoredUris: subscription.honoredFilter.resourceSubscriptions ?? [],
|
||||
});
|
||||
|
|
@ -5024,7 +5133,7 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
const shouldReconnect =
|
||||
reason === "remote" &&
|
||||
!isTerminalStatus(this.status) &&
|
||||
this.subscribedResources.size > 0;
|
||||
this.wantsModernStream();
|
||||
if (!shouldReconnect) {
|
||||
// "stream gone but subscriptions remain" renders the same whether we gave
|
||||
// up after failed reconnects or the server closed it gracefully: keep the
|
||||
|
|
@ -5033,7 +5142,7 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
// until the next `connect()` calls `resetSubscriptionStream`, which moves
|
||||
// them together — so "active with an empty set" is never observable.)
|
||||
this.setModernStreamState({
|
||||
active: this.subscribedResources.size > 0,
|
||||
active: this.modernStreamActive(),
|
||||
status: "ended",
|
||||
honoredUris: [],
|
||||
});
|
||||
|
|
@ -5052,10 +5161,11 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
* refresh owns the state as well as the filter — so both call sites gate on
|
||||
* the generation first.
|
||||
*
|
||||
* The empty case is the ordinary one: nothing subscribed, no stream, inactive.
|
||||
* The non-empty one exists because a failed re-listen leaves
|
||||
* `modernSubscription` null with URIs still subscribed, and nothing else will
|
||||
* notice: the reconnect machinery is reachable only from a stream that closed
|
||||
* The two branches are the empty and non-empty *filter* (#1920) — which is
|
||||
* "nothing subscribed" only when no list-change opt-in is live. The empty case
|
||||
* is the ordinary one: nothing to listen for, no stream, inactive. The
|
||||
* non-empty one exists because a failed re-listen leaves `modernSubscription`
|
||||
* null with the filter still wanting a stream, and nothing else will notice: the reconnect machinery is reachable only from a stream that closed
|
||||
* or a reconnect that failed, and neither happened here. Left alone, the state
|
||||
* keeps whatever the last success (or the optimistic `"connecting"`) wrote — a
|
||||
* badge that will never change over subscriptions the server may never have
|
||||
|
|
@ -5063,17 +5173,17 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
* Subscribe early-returns on the URI already being in the set).
|
||||
*
|
||||
* So it reconnects rather than settling for an honest-but-dead `"ended"`:
|
||||
* every other route to "stream gone, URIs live" either expects the close or
|
||||
* every other route to "stream gone, filter live" either expects the close or
|
||||
* has exhausted the retry cap, and this is the one that has made no attempt
|
||||
* at all. `scheduleModernReconnect` fits as-is — a user-initiated refresh
|
||||
* already reset `modernReconnectAttempts`, so it starts at the base delay; the
|
||||
* timer bails on a terminal status or an emptied set; and past the cap
|
||||
* timer bails on a terminal status or an emptied filter; and past the cap
|
||||
* `onModernReconnectFailed` lands on the same `"ended"` badge. The state
|
||||
* therefore becomes true or ends after a real attempt. The caller still sees
|
||||
* its error either way — the retry is about the subscriptions, not the call.
|
||||
*/
|
||||
private reconcileModernStreamStateAfterFailedRefresh(): void {
|
||||
if (this.subscribedResources.size === 0) {
|
||||
if (!this.wantsModernStream()) {
|
||||
this.setModernStreamState(INACTIVE_SUBSCRIPTION_STREAM_STATE);
|
||||
return;
|
||||
}
|
||||
|
|
@ -5088,7 +5198,7 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
*/
|
||||
private scheduleModernReconnect(): void {
|
||||
this.setModernStreamState({
|
||||
active: true,
|
||||
active: this.modernStreamActive(),
|
||||
status: "reconnecting",
|
||||
honoredUris: [],
|
||||
});
|
||||
|
|
@ -5101,10 +5211,7 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
this.modernReconnectTimer = undefined;
|
||||
// Disconnect/unsubscribe may have raced the timer — bail if the reconnect
|
||||
// is no longer wanted.
|
||||
if (
|
||||
isTerminalStatus(this.status) ||
|
||||
this.subscribedResources.size === 0
|
||||
) {
|
||||
if (isTerminalStatus(this.status) || !this.wantsModernStream()) {
|
||||
return;
|
||||
}
|
||||
this.refreshModernSubscription(true).catch(() =>
|
||||
|
|
@ -5123,10 +5230,10 @@ export class InspectorClient extends InspectorClientEventTarget {
|
|||
if (
|
||||
this.modernReconnectAttempts > MODERN_RECONNECT_MAX_ATTEMPTS ||
|
||||
isTerminalStatus(this.status) ||
|
||||
this.subscribedResources.size === 0
|
||||
!this.wantsModernStream()
|
||||
) {
|
||||
this.setModernStreamState({
|
||||
active: this.subscribedResources.size > 0,
|
||||
active: this.modernStreamActive(),
|
||||
status: "ended",
|
||||
honoredUris: [],
|
||||
});
|
||||
|
|
|
|||
|
|
@ -436,9 +436,14 @@ export type ResourceSubscriptionStreamStatus =
|
|||
* persistent stream, so `active` is `false` and the UI surfaces no stream chrome.
|
||||
* On the modern era all subscriptions are a filter over one long-lived
|
||||
* `subscriptions/listen` stream; `active` is `true` whenever that stream is being
|
||||
* managed (i.e. at least one URI is subscribed), and `honoredUris` is the subset
|
||||
* of requested URIs the server acknowledged in its `honoredFilter` (may be a
|
||||
* strict subset — a server is allowed to decline some).
|
||||
* managed *for resource subscriptions* (i.e. at least one URI is subscribed), and
|
||||
* `honoredUris` is the subset of requested URIs the server acknowledged in its
|
||||
* `honoredFilter` (may be a strict subset — a server is allowed to decline some).
|
||||
*
|
||||
* `active: false` does not imply no stream: the same stream also carries the
|
||||
* list-change opt-ins, so it can be open with no subscribed URI at all (#1920).
|
||||
* This state describes the Subscriptions section, which has nothing to show for
|
||||
* such a stream.
|
||||
*/
|
||||
export interface ResourceSubscriptionStreamState {
|
||||
active: boolean;
|
||||
|
|
|
|||
Loading…
Reference in a new issue