From eda8ec0db8afdf0543567019c0771bdf0d3f66bd Mon Sep 17 00:00:00 2001 From: cliffhall Date: Mon, 10 Aug 2026 23:34:19 -0400 Subject: [PATCH] fix: guard the connect-time listen failure on the refresh generation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot review: `openModernListenStreamOnConnect`'s catch reconciled unconditionally, unlike the subscribe/unsubscribe call sites. The `connect` event is dispatched before the stream opens, so a listener subscribing to a resource can start and acknowledge a newer refresh while this one is still awaiting its `listen()` — and reconciling anyway would arm a reconnect that tears down a healthy stream. Apply the same generation test. Also consolidates the tests' private-member access onto the file's existing `internals()` helper (one justified double cast, hoisted and documented) and makes the failure stub bump the generation the way the real method does. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01HDdo1rNRVQnVRRdsqSrvG3 --- .../inspectorClient-subscriptions-era.test.ts | 102 ++++++++++++------ core/mcp/inspectorClient.ts | 12 ++- 2 files changed, 80 insertions(+), 34 deletions(-) diff --git a/clients/web/src/test/integration/mcp/inspectorClient-subscriptions-era.test.ts b/clients/web/src/test/integration/mcp/inspectorClient-subscriptions-era.test.ts index 6d182817..91daa638 100644 --- a/clients/web/src/test/integration/mcp/inspectorClient-subscriptions-era.test.ts +++ b/clients/web/src/test/integration/mcp/inspectorClient-subscriptions-era.test.ts @@ -124,6 +124,34 @@ describe("resource subscriptions era fork (#1630)", () => { return params.notifications as Record | 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 }; + modernSubscription: McpSubscription | null; + modernListenGeneration: number; + modernReconnectAttempts: number; + subscribedResources: Set; + refreshModernSubscription(fromReconnect?: boolean): Promise; + 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({}); @@ -322,15 +350,7 @@ describe("resource subscriptions era fork (#1630)", () => { const started = await startToolsOnlyServer({ tools: true }); const { connected } = await connect(started.url, "modern"); - const int = connected as unknown as { - modernSubscription: McpSubscription | null; - modernListenGeneration: number; - onModernSubscriptionClosed( - subscription: McpSubscription, - reason: "local" | "graceful" | "remote", - generation: number, - ): void; - }; + const int = internals(connected); const dropped = int.modernSubscription; expect(dropped).not.toBeNull(); if (!dropped) return; @@ -367,13 +387,15 @@ describe("resource subscriptions era fork (#1630)", () => { ); // 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). - ( - connected as unknown as { - refreshModernSubscription: () => Promise; - } - ).refreshModernSubscription = () => - Promise.reject(new Error("listen boom")); + // `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; @@ -383,6 +405,37 @@ describe("resource subscriptions era fork (#1630)", () => { "reconnecting", ); }); + + it("leaves a superseded connect-time failure to the refresh that owns the stream", async () => { + // `connect` is dispatched before the stream is opened, so a listener can + // start and acknowledge a newer refresh while this one is still awaiting + // its `listen()`. Reconciling anyway would arm a reconnect against a + // healthy stream and tear it down — the same ownership test the + // subscribe/unsubscribe paths make. + 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", () => { @@ -433,23 +486,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 }; - modernSubscription: McpSubscription | null; - modernListenGeneration: number; - modernReconnectAttempts: number; - subscribedResources: Set; - 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; diff --git a/core/mcp/inspectorClient.ts b/core/mcp/inspectorClient.ts index 31a7e5e3..c9ec8207 100644 --- a/core/mcp/inspectorClient.ts +++ b/core/mcp/inspectorClient.ts @@ -4981,9 +4981,17 @@ export class InspectorClient extends InspectorClientEventTarget { * 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. This call is *not* the only refresh that can be in flight: + * the `connect` event has already been dispatched, so a listener subscribing + * to a resource can start and acknowledge a newer refresh while this one is + * still awaiting its `listen()`. Reconciling unconditionally would then arm a + * reconnect that tears down a stream that is up and healthy. */ private async openModernListenStreamOnConnect(): Promise { if (!this.isModernEra() || !this.wantsModernStream()) return; + const generationBefore = this.modernListenGeneration; try { await this.refreshModernSubscription(); } catch (error) { @@ -4991,7 +4999,9 @@ export class InspectorClient extends InspectorClientEventTarget { { error }, "Failed to open the modern subscriptions/listen stream on connect", ); - this.reconcileModernStreamStateAfterFailedRefresh(); + if (this.modernListenGeneration === generationBefore + 1) { + this.reconcileModernStreamStateAfterFailedRefresh(); + } } }