Fix SSE keep-alive timer lifecycle in Streamable HTTP server transport (v1.x) (#2547)

.changeset/sse-keepalive-timer-lifecycle.md

@@ -0,0 +1,8 @@
+---
+'@modelcontextprotocol/sdk': patch
+---
+
+Fix SSE keep-alive timer lifecycle in the Streamable HTTP server transport. Keep-alive timers are now owned by the stream they were armed for, so a stale disconnect arriving after a reconnect no longer stops the live stream's keep-alive, a resume closes the superseded stream
+cleanly, and a failed priming-event write no longer leaves a timer running or a stale request correlation behind. A resume that completes after the transport closed now gets a 404 instead of a silent stream nothing will ever write to. Non-finite `keepAliveMs` values now disable
+keep-alive, and values above 2^31-1 are clamped instead of firing every millisecond. With an event store configured, a response completing while its request stream is disconnected is now stored for `Last-Event-ID` replay and its request correlations released when the client can
+resume the stream, instead of `send()` rejecting; for clients that cannot resume (no priming event), `send()` still rejects so the failure surfaces. Replays no longer double-deliver events written concurrently with a resume, and requests racing transport close now receive a 404.

src/server/sseKeepAlive.ts

@@ -0,0 +1,15 @@
+/** Default interval between SSE keep-alive comment frames. */
+export const DEFAULT_SSE_KEEP_ALIVE_MS = 15_000;
+
+const MAX_TIMER_DELAY_MS = 2 ** 31 - 1;
+
+/** Arms an unref'd timer, or disables keep-alive for invalid delays. */
+export function armSseKeepAlive(intervalMs: number, onTick: () => void): ReturnType<typeof setInterval> | undefined {
+    if (!Number.isFinite(intervalMs) || intervalMs < 1) {
+        return undefined;
+    }
+
+    const timer = setInterval(onTick, Math.min(intervalMs, MAX_TIMER_DELAY_MS));
+    (timer as { unref?: () => void }).unref?.();
+    return timer;
+}

src/server/webStandardStreamableHttp.ts

@@ -10,6 +10,7 @@
 import { isJsonContentType } from '../shared/mediaType.js';
 import { Transport } from '../shared/transport.js';
 import { AuthInfo } from './auth/types.js';
+import { armSseKeepAlive, DEFAULT_SSE_KEEP_ALIVE_MS } from './sseKeepAlive.js';
 import {
     MessageExtraInfo,
     RequestInfo,
@@ -71,6 +72,12 @@ interface StreamMapping {
     resolveJson?: (response: Response) => void;
     /** Cleanup function to close stream and remove mapping */
     cleanup: () => void;
+    /**
+     * Event IDs delivered to this stream by replay when it was registered.
+     * Lets send() avoid re-delivering events the replay already wrote when a
+     * resume raced in-flight event-store writes.
+     */
+    replayedEventIds?: Set<string>;
 }
 
 /**
@@ -148,22 +155,12 @@ export interface WebStandardStreamableHTTPServerTransportOptions {
     retryInterval?: number;
 
     /**
-     * Interval in milliseconds between SSE keep-alive comment frames (`: keepalive`)
-     * written to open SSE streams. Keep-alive frames prevent idle streams (e.g. the
-     * standalone GET stream, or a POST stream during a long-running tool call) from
-     * being killed by intermediaries and server idle timeouts, which clients observe
-     * as `SSE stream disconnected: TypeError: terminated`.
-     *
-     * Comment frames are ignored by SSE parsers and never surface as messages.
-     * Defaults to 15000 (per the WHATWG SSE spec recommendation of roughly every
-     * 15 seconds). Set to 0 to disable keep-alive frames.
+     * Interval in milliseconds between SSE keep-alive comment frames.
+     * Defaults to `15000`; values below 1 (including `0`) disable keep-alive.
      */
     keepAliveMs?: number;
 }
 
-/** Default interval between SSE keep-alive comment frames. */
-const DEFAULT_KEEP_ALIVE_MS = 15_000;
-
 /**
  * Options for handling a request
  */
@@ -230,6 +227,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
     private _hasHandledRequest: boolean = false;
     private _streamMapping: Map<string, StreamMapping> = new Map();
     private _requestToStreamMapping: Map<RequestId, string> = new Map();
+    /**
+     * Request streams whose client holds a Last-Event-ID cursor (at least one
+     * id-bearing event was delivered on them), so a stored response can be
+     * replayed on resume.
+     */
+    private _resumableStreams: Set<string> = new Set();
     private _requestResponseMap: Map<RequestId, JSONRPCMessage> = new Map();
     private _initialized: boolean = false;
     private _enableJsonResponse: boolean = false;
@@ -242,7 +245,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
     private _enableDnsRebindingProtection: boolean;
     private _retryInterval?: number;
     private _keepAliveMs: number;
-    private _keepAliveTimers: Map<string, ReturnType<typeof setInterval>> = new Map();
+    private _closed = false;
 
     sessionId?: string;
     onclose?: () => void;
@@ -259,43 +262,31 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         this._allowedOrigins = options.allowedOrigins;
         this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false;
         this._retryInterval = options.retryInterval;
-        this._keepAliveMs = options.keepAliveMs ?? DEFAULT_KEEP_ALIVE_MS;
+        this._keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS;
     }
 
     /**
      * Arms a keep-alive interval for an SSE stream that periodically writes an SSE
      * comment frame so intermediaries and idle timeouts don't kill the connection.
-     * Replaces any timer already armed for the same stream id (a resumed stream
-     * re-registered under the same id supersedes its predecessor's timer). The
-     * timer is cleared via stopKeepAlive when the stream is cleaned up, and
-     * clears itself if a write fails (stream already closed/cancelled).
+     * The returned timer is owned by the stream it was armed for: the stream's
+     * cancel/cleanup callbacks must clear it. The interval clears itself if a
+     * write fails (stream already closed/cancelled).
      */
-    private startKeepAlive(streamId: string, controller: ReadableStreamDefaultController<Uint8Array>, encoder: TextEncoder): void {
-        if (this._keepAliveMs <= 0) {
-            return;
-        }
-        this.stopKeepAlive(streamId);
-        const timer = setInterval(() => {
+    private startKeepAlive(
+        controller: ReadableStreamDefaultController<Uint8Array>,
+        encoder: TextEncoder
+    ): ReturnType<typeof setInterval> | undefined {
+        // After close() nothing may arm: the stream cleanups have already run.
+        if (this._closed) return undefined;
+
+        const timer = armSseKeepAlive(this._keepAliveMs, () => {
             try {
                 controller.enqueue(encoder.encode(': keepalive\n\n'));
             } catch {
-                this.stopKeepAlive(streamId);
+                if (timer !== undefined) clearInterval(timer);
             }
-        }, this._keepAliveMs);
-        // Don't let the keep-alive timer hold the process open (Node.js only)
-        (timer as { unref?: () => void }).unref?.();
-        this._keepAliveTimers.set(streamId, timer);
-    }
-
-    /**
-     * Clears the keep-alive interval for a stream, if one is armed.
-     */
-    private stopKeepAlive(streamId: string): void {
-        const timer = this._keepAliveTimers.get(streamId);
-        if (timer !== undefined) {
-            clearInterval(timer);
-            this._keepAliveTimers.delete(streamId);
-        }
+        });
+        return timer;
     }
 
     /**
@@ -376,6 +367,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
      * Returns a Response object (Web Standard)
      */
     async handleRequest(req: Request, options?: HandleRequestOptions): Promise<Response> {
+        if (this._closed) {
+            return this.createJsonErrorResponse(404, -32001, 'Session not found');
+        }
+
         // In stateless mode (no sessionIdGenerator), each request must use a fresh transport.
         // Reusing a stateless transport causes message ID collisions between clients.
         if (!this.sessionIdGenerator && this._hasHandledRequest) {
@@ -430,6 +425,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
         }
         controller.enqueue(encoder.encode(primingEvent));
+        this._resumableStreams.add(streamId);
     }
 
     /**
@@ -472,6 +468,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
 
         const encoder = new TextEncoder();
         let streamController: ReadableStreamDefaultController<Uint8Array>;
+        let keepAliveTimer: ReturnType<typeof setInterval> | undefined = undefined;
 
         // Create a ReadableStream with a controller we can use to push SSE events
         const readable = new ReadableStream<Uint8Array>({
@@ -479,16 +476,23 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 streamController = controller;
             },
             cancel: () => {
-                // Stream was cancelled by client
-                this.stopKeepAlive(this._standaloneSseStreamId);
-                this._streamMapping.delete(this._standaloneSseStreamId);
+                // Stream was cancelled by client. Only tear down the mapping if it
+                // still points at this stream — a stale cancel arriving after a
+                // reconnect re-registered the id must not tear down the successor.
+                if (keepAliveTimer !== undefined) {
+                    clearInterval(keepAliveTimer);
+                }
+                if (this._streamMapping.get(this._standaloneSseStreamId)?.controller === streamController) {
+                    this._streamMapping.delete(this._standaloneSseStreamId);
+                }
             }
         });
 
         const headers: Record<string, string> = {
             'Content-Type': 'text/event-stream',
             'Cache-Control': 'no-cache, no-transform',
-            Connection: 'keep-alive'
+            Connection: 'keep-alive',
+            'X-Accel-Buffering': 'no'
         };
 
         // After initialization, always include the session ID if we have one
@@ -501,7 +505,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             controller: streamController!,
             encoder,
             cleanup: () => {
-                this.stopKeepAlive(this._standaloneSseStreamId);
+                if (keepAliveTimer !== undefined) {
+                    clearInterval(keepAliveTimer);
+                }
                 this._streamMapping.delete(this._standaloneSseStreamId);
                 try {
                     streamController!.close();
@@ -511,7 +517,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             }
         });
 
-        this.startKeepAlive(this._standaloneSseStreamId, streamController!, encoder);
+        keepAliveTimer = this.startKeepAlive(streamController!, encoder);
 
         return new Response(readable, { headers });
     }
@@ -547,7 +553,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             const headers: Record<string, string> = {
                 'Content-Type': 'text/event-stream',
                 'Cache-Control': 'no-cache, no-transform',
-                Connection: 'keep-alive'
+                Connection: 'keep-alive',
+                'X-Accel-Buffering': 'no'
             };
 
             if (this.sessionId !== undefined) {
@@ -557,19 +564,31 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             // Create a ReadableStream with controller for SSE
             const encoder = new TextEncoder();
             let streamController: ReadableStreamDefaultController<Uint8Array>;
+            let keepAliveTimer: ReturnType<typeof setInterval> | undefined = undefined;
+            let replayedStreamId: string | undefined = undefined;
+            let cancelled = false;
 
             const readable = new ReadableStream<Uint8Array>({
                 start: controller => {
                     streamController = controller;
                 },
                 cancel: () => {
-                    // Stream was cancelled by client
-                    // Cleanup will be handled by the mapping
+                    // Stream was cancelled by client. Only tear down the mapping if
+                    // it still points at this stream — a stale cancel must not tear
+                    // down a successor registered by a later resume.
+                    cancelled = true;
+                    if (keepAliveTimer !== undefined) {
+                        clearInterval(keepAliveTimer);
+                    }
+                    if (replayedStreamId !== undefined && this._streamMapping.get(replayedStreamId)?.controller === streamController) {
+                        this._streamMapping.delete(replayedStreamId);
+                    }
                 }
             });
 
             // Replay events - returns the streamId for backwards compatibility
-            const replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, {
+            const replayedEventIds = new Set<string>();
+            replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, {
                 send: async (eventId: string, message: JSONRPCMessage) => {
                     const success = this.writeSSEEvent(streamController!, encoder, message, eventId);
                     if (!success) {
@@ -579,16 +598,38 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                         } catch {
                             // Controller might already be closed
                         }
+                    } else {
+                        replayedEventIds.add(eventId);
                     }
                 }
             });
 
+            // close() may have run, or the stream may have been cancelled, while
+            // the event store was replaying: don't hand out or register a stream
+            // that nothing can ever clean up again.
+            if (this._closed || cancelled) {
+                try {
+                    streamController!.close();
+                } catch {
+                    // Controller might already be closed
+                }
+                return this.createJsonErrorResponse(404, -32001, 'Session not found');
+            }
+
+            // A reconnect re-registers the same stream id: close the predecessor
+            // stream cleanly (clearing its keep-alive timer) instead of leaving it
+            // registered nowhere with its timer still running.
+            this._streamMapping.get(replayedStreamId)?.cleanup();
+
             this._streamMapping.set(replayedStreamId, {
                 controller: streamController!,
                 encoder,
+                replayedEventIds,
                 cleanup: () => {
-                    this.stopKeepAlive(replayedStreamId);
-                    this._streamMapping.delete(replayedStreamId);
+                    if (keepAliveTimer !== undefined) {
+                        clearInterval(keepAliveTimer);
+                    }
+                    this._streamMapping.delete(replayedStreamId!);
                     try {
                         streamController!.close();
                     } catch {
@@ -597,7 +638,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 }
             });
 
-            this.startKeepAlive(replayedStreamId, streamController!, encoder);
+            // The resume itself proves the client holds a Last-Event-ID cursor
+            this._resumableStreams.add(replayedStreamId);
+
+            keepAliveTimer = this.startKeepAlive(streamController!, encoder);
 
             return new Response(readable, { headers });
         } catch (error) {
@@ -712,6 +756,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 return this.createJsonErrorResponse(400, -32700, 'Parse error: Invalid JSON-RPC message');
             }
 
+            // close() may have run while awaiting the request body above: don't
+            // initialize a session (or fire onsessioninitialized) on a transport
+            // whose cleanups have already run.
+            if (this._closed) {
+                return this.createJsonErrorResponse(404, -32001, 'Session not found');
+            }
+
             // Check if this is an initialization request
             // https://spec.modelcontextprotocol.io/specification/2025-03-26/basic/lifecycle/
             const isInitializationRequest = messages.some(isInitializeRequest);
@@ -750,6 +801,14 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 }
             }
 
+            // close() may have run while awaiting the request body or the session
+            // initialization callback above: don't register streams into maps the
+            // close sweep has already cleared, or hand out a stream that nothing
+            // will ever write to.
+            if (this._closed) {
+                return this.createJsonErrorResponse(404, -32001, 'Session not found');
+            }
+
             // check if it contains requests
             const hasRequests = messages.some(isJSONRPCRequest);
 
@@ -798,22 +857,30 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             // SSE streaming mode - use ReadableStream with controller for more reliable data pushing
             const encoder = new TextEncoder();
             let streamController: ReadableStreamDefaultController<Uint8Array>;
+            let keepAliveTimer: ReturnType<typeof setInterval> | undefined = undefined;
 
             const readable = new ReadableStream<Uint8Array>({
                 start: controller => {
                     streamController = controller;
                 },
                 cancel: () => {
-                    // Stream was cancelled by client
-                    this.stopKeepAlive(streamId);
-                    this._streamMapping.delete(streamId);
+                    // Stream was cancelled by client. Only tear down the mapping if
+                    // it still points at this stream — a stale cancel arriving after
+                    // a resume re-registered the id must not tear down the successor.
+                    if (keepAliveTimer !== undefined) {
+                        clearInterval(keepAliveTimer);
+                    }
+                    if (this._streamMapping.get(streamId)?.controller === streamController) {
+                        this._streamMapping.delete(streamId);
+                    }
                 }
             });
 
             const headers: Record<string, string> = {
                 'Content-Type': 'text/event-stream',
-                'Cache-Control': 'no-cache',
-                Connection: 'keep-alive'
+                'Cache-Control': 'no-cache, no-transform',
+                Connection: 'keep-alive',
+                'X-Accel-Buffering': 'no'
             };
 
             // After initialization, always include the session ID if we have one
@@ -829,7 +896,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                         controller: streamController!,
                         encoder,
                         cleanup: () => {
-                            this.stopKeepAlive(streamId);
+                            if (keepAliveTimer !== undefined) {
+                                clearInterval(keepAliveTimer);
+                            }
                             this._streamMapping.delete(streamId);
                             try {
                                 streamController!.close();
@@ -842,33 +911,52 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 }
             }
 
-            this.startKeepAlive(streamId, streamController!, encoder);
+            try {
+                // Write priming event if event store is configured (after mapping is set up)
+                await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion);
 
-            // Write priming event if event store is configured (after mapping is set up)
-            await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion);
+                // handle each message
+                for (const message of messages) {
+                    // Build closeSSEStream callback for requests when eventStore is configured
+                    // AND client supports resumability (protocol version >= 2025-11-25).
+                    // Old clients can't resume if the stream is closed early because they
+                    // didn't receive a priming event with an event ID.
+                    let closeSSEStream: (() => void) | undefined;
+                    let closeStandaloneSSEStream: (() => void) | undefined;
+                    if (isJSONRPCRequest(message) && this._eventStore && clientProtocolVersion >= '2025-11-25') {
+                        closeSSEStream = () => {
+                            this.closeSSEStream(message.id);
+                        };
+                        closeStandaloneSSEStream = () => {
+                            this.closeStandaloneSSEStream();
+                        };
+                    }
 
-            // handle each message
-            for (const message of messages) {
-                // Build closeSSEStream callback for requests when eventStore is configured
-                // AND client supports resumability (protocol version >= 2025-11-25).
-                // Old clients can't resume if the stream is closed early because they
-                // didn't receive a priming event with an event ID.
-                let closeSSEStream: (() => void) | undefined;
-                let closeStandaloneSSEStream: (() => void) | undefined;
-                if (isJSONRPCRequest(message) && this._eventStore && clientProtocolVersion >= '2025-11-25') {
-                    closeSSEStream = () => {
-                        this.closeSSEStream(message.id);
-                    };
-                    closeStandaloneSSEStream = () => {
-                        this.closeStandaloneSSEStream();
-                    };
+                    this.onmessage?.(message, { authInfo: options?.authInfo, requestInfo, closeSSEStream, closeStandaloneSSEStream });
                 }
-
-                this.onmessage?.(message, { authInfo: options?.authInfo, requestInfo, closeSSEStream, closeStandaloneSSEStream });
+            } catch (error) {
+                // Failing before the Response is handed out means nothing can ever
+                // cancel this stream: release it and its request correlations here.
+                this._streamMapping.get(streamId)?.cleanup();
+                this._resumableStreams.delete(streamId);
+                for (const message of messages) {
+                    if (isJSONRPCRequest(message)) {
+                        this._requestToStreamMapping.delete(message.id);
+                    }
+                }
+                throw error;
             }
             // The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses
             // This will be handled by the send() method when responses are ready
 
+            // Arm keep-alive only after the fallible awaits above: an error path
+            // returns 400 and discards the Response, so nothing could ever cancel
+            // the stream and clear an already-armed timer. Skip if the responses
+            // already completed and cleaned the stream up.
+            if (this._streamMapping.get(streamId)?.controller === streamController!) {
+                keepAliveTimer = this.startKeepAlive(streamController!, encoder);
+            }
+
             return new Response(readable, { status: 200, headers });
         } catch (error) {
             // return JSON-RPC formatted error
@@ -890,9 +978,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             return protocolError;
         }
 
-        await Promise.resolve(this._onsessionclosed?.(this.sessionId!));
-        await this.close();
-        return new Response(null, { status: 200 });
+        try {
+            await Promise.resolve(this._onsessionclosed?.(this.sessionId!));
+            return new Response(null, { status: 200 });
+        } finally {
+            await this.close();
+        }
     }
 
     /**
@@ -961,18 +1052,22 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
     }
 
     async close(): Promise<void> {
-        // Close all SSE connections
+        if (this._closed) {
+            return;
+        }
+        // Stop any deferred keep-alive arm (e.g. a replay await resolving after
+        // close) from creating a timer nothing can clear.
+        this._closed = true;
+
+        // Close all SSE connections (each cleanup clears its own keep-alive timer)
         this._streamMapping.forEach(({ cleanup }) => {
             cleanup();
         });
         this._streamMapping.clear();
 
-        // Clear any keep-alive timers not already cleared by stream cleanup
-        this._keepAliveTimers.forEach(timer => clearInterval(timer));
-        this._keepAliveTimers.clear();
-
         // Clear any pending responses
         this._requestResponseMap.clear();
+        this._resumableStreams.clear();
         this.onclose?.();
     }
 
@@ -1032,8 +1127,14 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 return;
             }
 
-            // Send the message to the standalone SSE stream
-            if (standaloneSse.controller && standaloneSse.encoder) {
+            // Send the message to the standalone SSE stream, unless this stream
+            // was registered by a replay that already delivered this event (the
+            // store made it visible before this call resumed).
+            if (
+                standaloneSse.controller &&
+                standaloneSse.encoder &&
+                (eventId === undefined || !standaloneSse.replayedEventIds?.has(eventId))
+            ) {
                 this.writeSSEEvent(standaloneSse.controller, standaloneSse.encoder, message, eventId);
             }
             return;
@@ -1045,17 +1146,31 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             throw new Error(`No connection established for request ID: ${String(requestId)}`);
         }
 
-        const stream = this._streamMapping.get(streamId);
+        let stream = this._streamMapping.get(streamId);
 
-        if (!this._enableJsonResponse && stream?.controller && stream?.encoder) {
-            // For SSE responses, generate event ID if event store is provided
+        if (!this._enableJsonResponse) {
+            // Store the event even if the stream is disconnected (e.g. after
+            // closeSSEStream switched the client to polling), so it can be
+            // replayed on reconnect — mirroring the standalone stream path.
             let eventId: string | undefined;
 
             if (this._eventStore) {
                 eventId = await this._eventStore.storeEvent(streamId, message);
+                // Re-fetch after the await: a resume completing meanwhile replaces
+                // this stream's registration with a successor, and the message
+                // must go to the live stream, not the evicted one.
+                stream = this._streamMapping.get(streamId);
+            }
+            // Skip the write if the stream was registered by a replay that
+            // already delivered this event (the store made it visible before
+            // this call resumed).
+            if (stream?.controller && stream?.encoder && (eventId === undefined || !stream.replayedEventIds?.has(eventId))) {
+                const written = this.writeSSEEvent(stream.controller, stream.encoder, message, eventId);
+                // The client now holds a Last-Event-ID cursor for this stream
+                if (written && eventId !== undefined) {
+                    this._resumableStreams.add(streamId);
+                }
             }
-            // Write the event to the response stream
-            this.writeSSEEvent(stream.controller, stream.encoder, message, eventId);
         }
 
         if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
@@ -1069,6 +1184,37 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
 
             if (allResponsesReady) {
                 if (!stream) {
+                    // close() ran while the event store write above was awaiting:
+                    // release the entries this call re-created after the sweep,
+                    // then treat the late completion as a no-op.
+                    if (this._closed) {
+                        for (const id of relatedIds) {
+                            this._requestResponseMap.delete(id);
+                            this._requestToStreamMapping.delete(id);
+                        }
+                        return;
+                    }
+                    if (!this._enableJsonResponse && this._eventStore && this._resumableStreams.has(streamId)) {
+                        // The stream went away (client disconnect, or
+                        // closeSSEStream switched the client to polling), but the
+                        // response was stored above and will be replayed on
+                        // resume — release the correlations rather than throwing.
+                        for (const id of relatedIds) {
+                            this._requestResponseMap.delete(id);
+                            this._requestToStreamMapping.delete(id);
+                        }
+                        this._resumableStreams.delete(streamId);
+                        return;
+                    }
+                    // Without resumability (or in JSON response mode, where the
+                    // pending HTTP response can never be settled by a replay),
+                    // a missing stream means the response cannot be delivered.
+                    // Release the correlations so the entries don't outlive the
+                    // undeliverable response.
+                    for (const id of relatedIds) {
+                        this._requestResponseMap.delete(id);
+                        this._requestToStreamMapping.delete(id);
+                    }
                     throw new Error(`No connection established for request ID: ${String(requestId)}`);
                 }
                 if (this._enableJsonResponse && stream.resolveJson) {
@@ -1096,6 +1242,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                     this._requestResponseMap.delete(id);
                     this._requestToStreamMapping.delete(id);
                 }
+                this._resumableStreams.delete(streamId);
             }
         }
     }

test/server/sseKeepAlive.test.ts

@@ -0,0 +1,29 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { armSseKeepAlive } from '../../src/server/sseKeepAlive.js';
+
+describe('armSseKeepAlive', () => {
+    beforeEach(() => vi.useFakeTimers());
+    afterEach(() => vi.useRealTimers());
+
+    it.each([0, -1, 0.5, Number.NaN, Number.POSITIVE_INFINITY])('disables invalid delay %s', delay => {
+        expect(armSseKeepAlive(delay, () => {})).toBeUndefined();
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('ticks at the configured interval', async () => {
+        const tick = vi.fn();
+        const timer = armSseKeepAlive(1_000, tick)!;
+        await vi.advanceTimersByTimeAsync(3_000);
+        expect(tick).toHaveBeenCalledTimes(3);
+        clearInterval(timer);
+    });
+
+    it('clamps overflowing delays instead of creating a 1ms timer', async () => {
+        const tick = vi.fn();
+        const timer = armSseKeepAlive(2 ** 31, tick)!;
+        await vi.advanceTimersByTimeAsync(60_000);
+        expect(tick).not.toHaveBeenCalled();
+        clearInterval(timer);
+    });
+});

test/server/streamableHttp.test.ts

@@ -3412,9 +3412,9 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => {
     it('should supersede the previous keep-alive timer when a replayed stream re-registers under the same stream id', async () => {
         // Event store WITHOUT the optional getStreamIdForEventId — the replay
         // path then skips its 409 conflict check, so a reconnect re-registers
-        // the same stream id. The predecessor's timer must be replaced, not
-        // orphaned (an orphaned timer's failing write would clear the live
-        // stream's keep-alive via stopKeepAlive on the shared stream id).
+        // the same stream id. The predecessor stream must be closed and its
+        // timer cleared, not left orphaned; each timer is owned by its stream,
+        // so a stale timer's failing write only ever clears itself.
         const eventStore: EventStore = {
             async storeEvent(): Promise<EventId> {
                 return 'evt-1';
@@ -3448,3 +3448,879 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => {
         await transport.close();
     });
 });
+
+describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () => {
+    /** Shorthand to build a Web Standard Request for direct transport testing. */
+    function req(method: string, opts?: { body?: unknown; headers?: Record<string, string> }): Request {
+        const headers: Record<string, string> = { ...opts?.headers };
+        if (method === 'POST') {
+            headers['Accept'] ??= 'application/json, text/event-stream';
+            headers['Content-Type'] ??= 'application/json';
+        } else if (method === 'GET') {
+            headers['Accept'] ??= 'text/event-stream';
+        }
+        return new Request('http://localhost/mcp', {
+            method,
+            headers,
+            body: opts?.body !== undefined ? JSON.stringify(opts.body) : undefined
+        });
+    }
+
+    /**
+     * Minimal event store WITHOUT the optional getStreamIdForEventId, so the
+     * replay path re-registers a stream id without a 409 conflict check —
+     * the reconnect shape these tests exercise.
+     */
+    function createSimpleEventStore(): EventStore {
+        const events: { id: EventId; streamId: StreamId; message: JSONRPCMessage }[] = [];
+        let counter = 0;
+        return {
+            async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId> {
+                const id = `${streamId}#${counter++}`;
+                events.push({ id, streamId, message });
+                return id;
+            },
+            async replayEventsAfter(lastEventId: EventId, { send }): Promise<StreamId> {
+                const index = events.findIndex(e => e.id === lastEventId);
+                const streamId = events[index]?.streamId ?? '_GET_stream';
+                for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) {
+                    await send(event.id, event.message);
+                }
+                return streamId;
+            }
+        };
+    }
+
+    async function createTransport(options?: {
+        keepAliveMs?: number;
+        eventStore?: EventStore;
+    }): Promise<{ transport: WebStandardStreamableHTTPServerTransport; sessionId: string }> {
+        const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), ...options });
+        await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport);
+        const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        expect(initResponse.status).toBe(200);
+        return { transport, sessionId: initResponse.headers.get('mcp-session-id') as string };
+    }
+
+    function get(sessionId: string, lastEventId?: string): Request {
+        const headers: Record<string, string> = { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' };
+        if (lastEventId !== undefined) {
+            headers['Last-Event-ID'] = lastEventId;
+        }
+        return req('GET', { headers });
+    }
+
+    /** Opens the standalone GET stream and returns a real stored event id to resume from. */
+    async function openGetStreamWithEvent(
+        transport: WebStandardStreamableHTTPServerTransport,
+        sessionId: string
+    ): Promise<{ reader: ReadableStreamDefaultReader<Uint8Array>; eventId: string }> {
+        const response = await transport.handleRequest(get(sessionId));
+        expect(response.status).toBe(200);
+        const reader = response.body!.getReader();
+        await transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'x' } });
+        const { value } = await reader.read();
+        const eventId = /^id: (.+)$/m.exec(new TextDecoder().decode(value))?.[1];
+        expect(eventId).toBeDefined();
+        return { reader, eventId: eventId! };
+    }
+
+    beforeEach(() => {
+        vi.useFakeTimers();
+    });
+
+    afterEach(() => {
+        vi.useRealTimers();
+    });
+
+    it('should keep the resumed stream alive when the predecessor connection is cancelled late', async () => {
+        const { transport, sessionId } = await createTransport({ eventStore: createSimpleEventStore() });
+        const { reader: staleReader, eventId } = await openGetStreamWithEvent(transport, sessionId);
+
+        // Client reconnects and resumes while the old connection is still half-open
+        const resumed = await transport.handleRequest(get(sessionId, eventId));
+        expect(resumed.status).toBe(200);
+        const resumedReader = resumed.body!.getReader();
+
+        // The old connection's socket finally dies. This must not tear down the
+        // resumed stream's keep-alive timer or its stream registration.
+        await staleReader.cancel();
+
+        await vi.advanceTimersByTimeAsync(15000);
+        const { value: frame } = await resumedReader.read();
+        expect(new TextDecoder().decode(frame)).toBe(': keepalive\n\n');
+
+        // Server-to-client messages must still reach the resumed stream too
+        await transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'after' } });
+        const { value: notification } = await resumedReader.read();
+        expect(new TextDecoder().decode(notification)).toContain('notifications/message');
+
+        await transport.close();
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should close the predecessor stream when a resume re-registers its stream id', async () => {
+        const { transport, sessionId } = await createTransport({ eventStore: createSimpleEventStore() });
+        const { reader: staleReader, eventId } = await openGetStreamWithEvent(transport, sessionId);
+
+        await transport.handleRequest(get(sessionId, eventId));
+
+        // The superseded stream must end cleanly rather than hang as a zombie
+        const { done } = await staleReader.read();
+        expect(done).toBe(true);
+
+        await transport.close();
+    });
+
+    it('should not arm keep-alive when the transport closes during an event-store replay await', async () => {
+        let releaseReplay: (() => void) | undefined;
+        const eventStore: EventStore = {
+            async storeEvent(): Promise<EventId> {
+                return 'evt-1';
+            },
+            async replayEventsAfter(): Promise<StreamId> {
+                await new Promise<void>(resolve => {
+                    releaseReplay = resolve;
+                });
+                return '_GET_stream';
+            }
+        };
+        const { transport, sessionId } = await createTransport({ eventStore });
+
+        // Enter replayEvents and park on the replayEventsAfter await
+        const pendingGet = transport.handleRequest(get(sessionId, 'evt-1'));
+        await vi.advanceTimersByTimeAsync(0);
+        expect(releaseReplay).toBeDefined();
+
+        // Close the transport mid-await, then let the replay continuation run.
+        // The deferred continuation must not arm a timer close() can never clear,
+        // and must not hand out a 200 SSE stream nothing will ever write to.
+        await transport.close();
+        releaseReplay!();
+        const response = await pendingGet;
+
+        expect(response.status).toBe(404);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should not leave a keep-alive timer armed when the priming event write fails', async () => {
+        let failStore = false;
+        const eventStore: EventStore = {
+            async storeEvent(streamId: StreamId): Promise<EventId> {
+                if (failStore) {
+                    throw new Error('store unavailable');
+                }
+                return `${streamId}#0`;
+            },
+            async replayEventsAfter(): Promise<StreamId> {
+                return '_GET_stream';
+            }
+        };
+        const { transport, sessionId } = await createTransport({ eventStore });
+        // Let the init response complete so its own stream cleanup has run
+        await vi.advanceTimersByTimeAsync(0);
+        expect(vi.getTimerCount()).toBe(0);
+
+        failStore = true;
+        const response = await transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/list', params: {}, id: 'req-1' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
+            })
+        );
+
+        // The 400 discards the Response, so nothing could ever cancel the stream —
+        // no timer may be left behind.
+        expect(response.status).toBe(400);
+        expect(vi.getTimerCount()).toBe(0);
+
+        // The failed request's stream and correlation must be released too: a
+        // late response for it has nowhere to go and must say so, rather than
+        // being written to a dead stream.
+        await expect(transport.send({ jsonrpc: '2.0', id: 'req-1', result: { tools: [] } })).rejects.toThrow(
+            'No connection established for request ID'
+        );
+
+        await transport.close();
+    });
+
+    it.each([NaN, Infinity])('should disable keep-alive for non-finite keepAliveMs (%s)', async keepAliveMs => {
+        const { transport, sessionId } = await createTransport({ keepAliveMs });
+
+        const response = await transport.handleRequest(get(sessionId));
+        expect(response.status).toBe(200);
+
+        // A non-finite interval must disable keep-alive, not arm a broken timer
+        // (setInterval would clamp it to ~1ms and flood the stream).
+        expect(vi.getTimerCount()).toBe(0);
+
+        await transport.close();
+    });
+
+    it('should clamp keepAliveMs above 2^31-1 instead of flooding the stream', async () => {
+        const { transport, sessionId } = await createTransport({ keepAliveMs: 2 ** 31 });
+
+        const response = await transport.handleRequest(get(sessionId));
+        const reader = response.body!.getReader();
+
+        // Un-clamped, setInterval treats 2^31 as ~1ms and floods; clamped, no
+        // frame is due for a very long time.
+        await vi.advanceTimersByTimeAsync(60000);
+        const read = reader.read();
+        const raced = await Promise.race([read, Promise.resolve('pending')]);
+        expect(raced).toBe('pending');
+
+        await transport.close();
+    });
+
+    it('should deliver a response to the successor stream when a resume completes during the event-store write', async () => {
+        // storeEvent parks on the RESPONSE write, so a resume can complete and
+        // replace the stream registration while send() is awaiting.
+        const events: { id: string; streamId: string; message: JSONRPCMessage }[] = [];
+        let counter = 0;
+        let parkNext = false;
+        let releaseStore: (() => void) | undefined;
+        const eventStore: EventStore = {
+            async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId> {
+                if (parkNext) {
+                    parkNext = false;
+                    await new Promise<void>(resolve => {
+                        releaseStore = resolve;
+                    });
+                }
+                const id = `${streamId}#${counter++}`;
+                events.push({ id, streamId, message });
+                return id;
+            },
+            async replayEventsAfter(lastEventId: EventId, { send }): Promise<StreamId> {
+                const index = events.findIndex(e => e.id === lastEventId);
+                const streamId = events[index]?.streamId ?? '_GET_stream';
+                for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) {
+                    await send(event.id, event.message);
+                }
+                return streamId;
+            }
+        };
+
+        const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore });
+        const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+        let resolveTool: (() => void) | undefined;
+        mcpServer.tool('slow', async () => {
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+        const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        // Original request stream; capture its priming event id for the resume
+        const original = await transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
+            })
+        );
+        const originalReader = original.body!.getReader();
+        const { value: priming } = await originalReader.read();
+        const primingEventId = /^id: (.+)$/m.exec(new TextDecoder().decode(priming))?.[1];
+        expect(primingEventId).toBeDefined();
+
+        // Complete the tool; the response's storeEvent parks inside send()
+        parkNext = true;
+        resolveTool?.();
+        await vi.advanceTimersByTimeAsync(0);
+        expect(releaseStore).toBeDefined();
+
+        // Client reconnects and resumes the request stream while send() is parked
+        const resumed = await transport.handleRequest(
+            req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': primingEventId! } })
+        );
+        expect(resumed.status).toBe(200);
+        const resumedReader = resumed.body!.getReader();
+
+        // Release the parked write: the response must reach the live (resumed)
+        // stream, not vanish into the evicted one.
+        releaseStore!();
+        await vi.advanceTimersByTimeAsync(0);
+
+        let resumedData = '';
+        for (let i = 0; i < 3 && !resumedData.includes('call-1'); i++) {
+            const { value, done } = await resumedReader.read();
+            if (done) {
+                break;
+            }
+            resumedData += new TextDecoder().decode(value);
+        }
+        expect(resumedData).toContain('"id":"call-1"');
+        expect(resumedData).toContain('done');
+
+        await transport.close();
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should not register streams when the transport closes during the session initialization callback', async () => {
+        let releaseInit: (() => void) | undefined;
+        const transport = new WebStandardStreamableHTTPServerTransport({
+            sessionIdGenerator: () => randomUUID(),
+            onsessioninitialized: () =>
+                new Promise<void>(resolve => {
+                    releaseInit = resolve;
+                })
+        });
+        await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport);
+
+        // Initialization parks on the onsessioninitialized await
+        const pendingInit = transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        await vi.advanceTimersByTimeAsync(0);
+        expect(releaseInit).toBeDefined();
+
+        // Close mid-await, then let the continuation run: it must not hand out
+        // a 200 SSE stream nothing will ever write to, nor arm a timer.
+        await transport.close();
+        releaseInit!();
+        const response = await pendingInit;
+
+        expect(response.status).toBe(404);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should not double-deliver a response when the event becomes replayable before the store write resolves', async () => {
+        // storeEvent persists the event (making it replay-visible) and THEN
+        // parks: a resume in that window replays the response to the successor,
+        // and the parked send() continuation must not write it a second time.
+        const events: { id: string; streamId: string; message: JSONRPCMessage }[] = [];
+        let counter = 0;
+        let parkNext = false;
+        let releaseStore: (() => void) | undefined;
+        const eventStore: EventStore = {
+            async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId> {
+                const id = `${streamId}#${counter++}`;
+                events.push({ id, streamId, message });
+                if (parkNext) {
+                    parkNext = false;
+                    await new Promise<void>(resolve => {
+                        releaseStore = resolve;
+                    });
+                }
+                return id;
+            },
+            async replayEventsAfter(lastEventId: EventId, { send }): Promise<StreamId> {
+                const index = events.findIndex(e => e.id === lastEventId);
+                const streamId = events[index]?.streamId ?? '_GET_stream';
+                for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) {
+                    await send(event.id, event.message);
+                }
+                return streamId;
+            }
+        };
+
+        const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore });
+        const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+        let resolveTool: (() => void) | undefined;
+        mcpServer.tool('slow', async () => {
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+        const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        const original = await transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
+            })
+        );
+        const originalReader = original.body!.getReader();
+        const { value: priming } = await originalReader.read();
+        const primingEventId = /^id: (.+)$/m.exec(new TextDecoder().decode(priming))?.[1];
+        expect(primingEventId).toBeDefined();
+
+        parkNext = true;
+        resolveTool?.();
+        await vi.advanceTimersByTimeAsync(0);
+        expect(releaseStore).toBeDefined();
+
+        // The response event is already visible in the store: the resume's
+        // replay delivers it to the successor stream.
+        const resumed = await transport.handleRequest(
+            req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': primingEventId! } })
+        );
+        const resumedReader = resumed.body!.getReader();
+
+        releaseStore!();
+        await vi.advanceTimersByTimeAsync(0);
+
+        let resumedData = '';
+        for (let i = 0; i < 4; i++) {
+            const { value, done } = await resumedReader.read();
+            if (done) {
+                break;
+            }
+            resumedData += new TextDecoder().decode(value);
+        }
+        const deliveries = resumedData.match(/"id":"call-1"/g) ?? [];
+        expect(deliveries).toHaveLength(1);
+
+        await transport.close();
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should not initialize a session when the transport closes while the request body is being read', async () => {
+        const onsessioninitialized = vi.fn();
+        const transport = new WebStandardStreamableHTTPServerTransport({
+            sessionIdGenerator: () => randomUUID(),
+            onsessioninitialized
+        });
+        await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport);
+
+        // Init request whose body stream parks until we release it
+        let releaseBody: (() => void) | undefined;
+        const encoder = new TextEncoder();
+        const body = new ReadableStream<Uint8Array>({
+            start: controller => {
+                releaseBody = () => {
+                    controller.enqueue(encoder.encode(JSON.stringify(TEST_MESSAGES.initialize)));
+                    controller.close();
+                };
+            }
+        });
+        const request = new Request('http://localhost/mcp', {
+            method: 'POST',
+            headers: { Accept: 'application/json, text/event-stream', 'Content-Type': 'application/json' },
+            body,
+            // @ts-expect-error duplex is required for streaming bodies but not yet in lib types
+            duplex: 'half'
+        });
+
+        const pending = transport.handleRequest(request);
+        await vi.advanceTimersByTimeAsync(0);
+        expect(releaseBody).toBeDefined();
+
+        // Close while req.json() is still reading, then deliver the body: the
+        // continuation must not initialize a session on the closed transport.
+        await transport.close();
+        releaseBody!();
+        const response = await pending;
+
+        expect(response.status).toBe(404);
+        expect(onsessioninitialized).not.toHaveBeenCalled();
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should store and replay a response completed after closeSSEStream switched the client to polling', async () => {
+        const events: { id: string; streamId: string; message: JSONRPCMessage }[] = [];
+        let counter = 0;
+        const eventStore: EventStore = {
+            async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId> {
+                const id = `${streamId}#${counter++}`;
+                events.push({ id, streamId, message });
+                return id;
+            },
+            async replayEventsAfter(lastEventId: EventId, { send }): Promise<StreamId> {
+                const index = events.findIndex(e => e.id === lastEventId);
+                const streamId = events[index]?.streamId ?? '_GET_stream';
+                for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) {
+                    await send(event.id, event.message);
+                }
+                return streamId;
+            }
+        };
+        const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore });
+        const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+        let resolveTool: (() => void) | undefined;
+        mcpServer.tool('slow', async () => {
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+        const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        const original = await transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
+            })
+        );
+        const { value: priming } = await original.body!.getReader().read();
+        const primingEventId = /^id: (.+)$/m.exec(new TextDecoder().decode(priming))?.[1];
+        expect(primingEventId).toBeDefined();
+
+        // Server switches the client to polling; the request stream is gone
+        transport.closeSSEStream('call-1');
+
+        // Tool completes with no stream attached: the response must be stored
+        // for replay, and the released correlations must not make send() throw.
+        resolveTool?.();
+        await vi.advanceTimersByTimeAsync(0);
+
+        // Client polls back in with Last-Event-ID and must receive the response
+        const resumed = await transport.handleRequest(
+            req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': primingEventId! } })
+        );
+        expect(resumed.status).toBe(200);
+        const resumedReader = resumed.body!.getReader();
+        let resumedData = '';
+        for (let i = 0; i < 3 && !resumedData.includes('call-1'); i++) {
+            const { value, done } = await resumedReader.read();
+            if (done) {
+                break;
+            }
+            resumedData += new TextDecoder().decode(value);
+        }
+        expect(resumedData).toContain('"id":"call-1"');
+        expect(resumedData).toContain('done');
+
+        await transport.close();
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should not re-deliver replayed server notifications to a resumed standalone stream', async () => {
+        const events: { id: string; streamId: string; message: JSONRPCMessage }[] = [];
+        let counter = 0;
+        let parkNext = false;
+        let releaseStore: (() => void) | undefined;
+        const eventStore: EventStore = {
+            async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId> {
+                const id = `${streamId}#${counter++}`;
+                events.push({ id, streamId, message });
+                if (parkNext) {
+                    parkNext = false;
+                    await new Promise<void>(resolve => {
+                        releaseStore = resolve;
+                    });
+                }
+                return id;
+            },
+            async replayEventsAfter(lastEventId: EventId, { send }): Promise<StreamId> {
+                const index = events.findIndex(e => e.id === lastEventId);
+                const streamId = events[index]?.streamId ?? '_GET_stream';
+                for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) {
+                    await send(event.id, event.message);
+                }
+                return streamId;
+            }
+        };
+        const { transport, sessionId } = await createTransport({ eventStore });
+
+        // First notification anchors the resume point
+        const first = await transport.handleRequest(get(sessionId));
+        const firstReader = first.body!.getReader();
+        await transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'anchor' } });
+        const { value } = await firstReader.read();
+        const anchorId = /^id: (.+)$/m.exec(new TextDecoder().decode(value))?.[1];
+        expect(anchorId).toBeDefined();
+
+        // Second notification: stored (replay-visible), then the write parks
+        parkNext = true;
+        const parkedSend = transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'raced' } });
+        await vi.advanceTimersByTimeAsync(0);
+        expect(releaseStore).toBeDefined();
+
+        // The old connection drops and the client resumes: replay delivers the
+        // second notification to the successor stream.
+        await firstReader.cancel();
+        const resumed = await transport.handleRequest(get(sessionId, anchorId));
+        const resumedReader = resumed.body!.getReader();
+
+        releaseStore!();
+        await parkedSend;
+
+        let resumedData = '';
+        for (let i = 0; i < 3; i++) {
+            const read = resumedReader.read();
+            const raced = await Promise.race([read, Promise.resolve('pending')]);
+            if (raced === 'pending') {
+                break;
+            }
+            const { value: chunk, done } = raced as ReadableStreamReadResult<Uint8Array>;
+            if (done) {
+                break;
+            }
+            resumedData += new TextDecoder().decode(chunk);
+        }
+        const deliveries = resumedData.match(/raced/g) ?? [];
+        expect(deliveries).toHaveLength(1);
+
+        await transport.close();
+    });
+
+    it('should fail loudly when a JSON-mode response completes after its stream is gone', async () => {
+        const transport = new WebStandardStreamableHTTPServerTransport({
+            sessionIdGenerator: () => randomUUID(),
+            enableJsonResponse: true
+        });
+        const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+        let resolveTool: (() => void) | undefined;
+        mcpServer.tool('slow', async () => {
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+        const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        const errors: Error[] = [];
+        mcpServer.server.onerror = error => {
+            errors.push(error);
+        };
+
+        void transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
+            })
+        );
+        await vi.advanceTimersByTimeAsync(0);
+
+        // The stream mapping disappears while the tool is still running
+        transport.closeSSEStream('call-1');
+        resolveTool?.();
+        await vi.advanceTimersByTimeAsync(0);
+
+        // In JSON mode nothing can ever replay the response: completing against
+        // a missing stream must surface an error, not vanish silently.
+        expect(errors.map(e => e.message).join('\n')).toContain('No connection established for request ID');
+
+        await transport.close();
+    });
+
+    it('should close the transport when the onsessionclosed callback throws on DELETE', async () => {
+        const transport = new WebStandardStreamableHTTPServerTransport({
+            sessionIdGenerator: () => randomUUID(),
+            onsessionclosed: () => {
+                throw new Error('session registry unavailable');
+            }
+        });
+        await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport);
+        const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        // Open the standalone stream so a keep-alive timer is armed
+        const getResponse = await transport.handleRequest(get(sessionId));
+        expect(getResponse.status).toBe(200);
+        expect(vi.getTimerCount()).toBe(1);
+
+        await expect(
+            transport.handleRequest(
+                new Request('http://localhost/mcp', {
+                    method: 'DELETE',
+                    headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
+                })
+            )
+        ).rejects.toThrow('session registry unavailable');
+
+        // The callback threw, but the transport must still have been closed:
+        // timers swept and subsequent requests rejected.
+        expect(vi.getTimerCount()).toBe(0);
+        const after = await transport.handleRequest(get(sessionId));
+        expect(after.status).toBe(404);
+    });
+
+    it('should reject requests with 404 after the transport is closed', async () => {
+        const { transport, sessionId } = await createTransport();
+        await transport.close();
+
+        const response = await transport.handleRequest(get(sessionId));
+        expect(response.status).toBe(404);
+    });
+
+    it('should surface an error when a response completes for a disconnected client that cannot resume', async () => {
+        // Pre-2025-11-25 clients never receive a priming event, so they cannot
+        // resume a request stream: a response completing after their disconnect
+        // is undeliverable and must surface, not be silently handed to replay.
+        const eventStore = createSimpleEventStore();
+        const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore });
+        const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+        let resolveTool: (() => void) | undefined;
+        mcpServer.tool('slow', async () => {
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+        const initResponse = await transport.handleRequest(
+            req('POST', {
+                body: {
+                    jsonrpc: '2.0',
+                    id: 1,
+                    method: 'initialize',
+                    params: { protocolVersion: '2025-03-26', capabilities: {}, clientInfo: { name: 'legacy', version: '1.0' } }
+                }
+            })
+        );
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        const errors: Error[] = [];
+        mcpServer.server.onerror = error => {
+            errors.push(error);
+        };
+
+        const response = await transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'legacy-1' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-03-26' }
+            })
+        );
+        expect(response.status).toBe(200);
+
+        // Client disconnects mid-call; no priming event was ever written, so
+        // no Last-Event-ID cursor exists for a resume.
+        await response.body!.getReader().cancel();
+        resolveTool?.();
+        await vi.advanceTimersByTimeAsync(0);
+
+        expect(errors.map(e => e.message).join('\n')).toContain('No connection established for request ID');
+
+        await transport.close();
+    });
+
+    it('should hand off to replay for a legacy client that received an id-bearing notification', async () => {
+        // A pre-2025-11-25 client gets no priming event, but any stored
+        // notification delivered on the stream carries an id — that cursor is
+        // enough to resume, so the completed response must be released to
+        // replay, not surfaced as an error.
+        const eventStore = createSimpleEventStore();
+        const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore });
+        const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: { logging: {} } });
+        let resolveTool: (() => void) | undefined;
+        mcpServer.tool('notifying', async (extra): Promise<CallToolResult> => {
+            await extra.sendNotification({ method: 'notifications/message', params: { level: 'info', data: 'progress' } });
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+        const initResponse = await transport.handleRequest(
+            req('POST', {
+                body: {
+                    jsonrpc: '2.0',
+                    id: 1,
+                    method: 'initialize',
+                    params: { protocolVersion: '2025-03-26', capabilities: {}, clientInfo: { name: 'legacy', version: '1.0' } }
+                }
+            })
+        );
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        const errors: Error[] = [];
+        mcpServer.server.onerror = error => {
+            errors.push(error);
+        };
+
+        const response = await transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'notifying', arguments: {} }, id: 'legacy-2' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-03-26' }
+            })
+        );
+        const reader = response.body!.getReader();
+        await vi.advanceTimersByTimeAsync(0);
+        const { value } = await reader.read();
+        const cursor = /^id: (.+)$/m.exec(new TextDecoder().decode(value))?.[1];
+        expect(cursor).toBeDefined();
+
+        // Client disconnects holding the notification's event id, then the tool completes
+        await reader.cancel();
+        resolveTool?.();
+        await vi.advanceTimersByTimeAsync(0);
+
+        expect(errors).toHaveLength(0);
+
+        // The resume from that cursor must replay the response
+        const resumed = await transport.handleRequest(get(sessionId, cursor!));
+        const resumedReader = resumed.body!.getReader();
+        let resumedData = '';
+        for (let i = 0; i < 3 && !resumedData.includes('legacy-2'); i++) {
+            const { value: chunk, done } = await resumedReader.read();
+            if (done) {
+                break;
+            }
+            resumedData += new TextDecoder().decode(chunk);
+        }
+        expect(resumedData).toContain('"id":"legacy-2"');
+
+        await transport.close();
+    });
+
+    it('should not surface an error when the transport closes during the response store write', async () => {
+        let parkNext = false;
+        let releaseStore: (() => void) | undefined;
+        const inner = createSimpleEventStore();
+        const eventStore: EventStore = {
+            async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId> {
+                if (parkNext) {
+                    parkNext = false;
+                    await new Promise<void>(resolve => {
+                        releaseStore = resolve;
+                    });
+                }
+                return inner.storeEvent(streamId, message);
+            },
+            replayEventsAfter: inner.replayEventsAfter.bind(inner)
+        };
+        const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore });
+        const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+        let resolveTool: (() => void) | undefined;
+        mcpServer.tool('slow', async () => {
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+        const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        const errors: Error[] = [];
+        mcpServer.server.onerror = error => {
+            errors.push(error);
+        };
+
+        void transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'race-1' },
+                headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
+            })
+        );
+        await vi.advanceTimersByTimeAsync(0);
+
+        // The response send parks inside storeEvent; close() sweeps everything
+        parkNext = true;
+        resolveTool?.();
+        await vi.advanceTimersByTimeAsync(0);
+        expect(releaseStore).toBeDefined();
+        await transport.close();
+        releaseStore!();
+        await vi.advanceTimersByTimeAsync(0);
+
+        // The transport is gone; the late completion must be a no-op, not a
+        // spurious 'No connection established' error.
+        expect(errors.filter(e => e.message.includes('No connection established'))).toHaveLength(0);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('should make close() idempotent', async () => {
+        const { transport } = await createTransport();
+        let oncloseCalls = 0;
+        transport.onclose = () => {
+            oncloseCalls++;
+        };
+
+        await transport.close();
+        await transport.close();
+
+        expect(oncloseCalls).toBe(1);
+    });
+});