diff --git a/.changeset/sse-keepalive-timer-lifecycle.md b/.changeset/sse-keepalive-timer-lifecycle.md new file mode 100644 index 0000000000..030f1ec4fe --- /dev/null +++ b/.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. diff --git a/src/server/sseKeepAlive.ts b/src/server/sseKeepAlive.ts new file mode 100644 index 0000000000..25df11871c --- /dev/null +++ b/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 | 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; +} diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 932ad56600..db83c7cf22 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/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; } /** @@ -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 = new Map(); private _requestToStreamMapping: Map = 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 = new Set(); private _requestResponseMap: Map = 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> = 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, encoder: TextEncoder): void { - if (this._keepAliveMs <= 0) { - return; - } - this.stopKeepAlive(streamId); - const timer = setInterval(() => { + private startKeepAlive( + controller: ReadableStreamDefaultController, + encoder: TextEncoder + ): ReturnType | 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 { + 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; + let keepAliveTimer: ReturnType | undefined = undefined; // Create a ReadableStream with a controller we can use to push SSE events const readable = new ReadableStream({ @@ -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 = { '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 = { '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; + let keepAliveTimer: ReturnType | undefined = undefined; + let replayedStreamId: string | undefined = undefined; + let cancelled = false; const readable = new ReadableStream({ 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(); + 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; + let keepAliveTimer: ReturnType | undefined = undefined; const readable = new ReadableStream({ 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 = { '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 { - // 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); } } } diff --git a/test/server/sseKeepAlive.test.ts b/test/server/sseKeepAlive.test.ts new file mode 100644 index 0000000000..238ac2f81c --- /dev/null +++ b/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); + }); +}); diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 2046a71697..99a6952088 100644 --- a/test/server/streamableHttp.test.ts +++ b/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 { 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 }): Request { + const headers: Record = { ...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 { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + 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 = { '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; 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 { + return 'evt-1'; + }, + async replayEventsAfter(): Promise { + await new Promise(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 { + if (failStore) { + throw new Error('store unavailable'); + } + return `${streamId}#0`; + }, + async replayEventsAfter(): Promise { + 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 { + if (parkNext) { + parkNext = false; + await new Promise(resolve => { + releaseStore = resolve; + }); + } + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + 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(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(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 { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + if (parkNext) { + parkNext = false; + await new Promise(resolve => { + releaseStore = resolve; + }); + } + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + 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(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({ + 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 { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + 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(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 { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + if (parkNext) { + parkNext = false; + await new Promise(resolve => { + releaseStore = resolve; + }); + } + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + 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; + 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(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(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 => { + await extra.sendNotification({ method: 'notifications/message', params: { level: 'info', data: 'progress' } }); + await new Promise(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 { + if (parkNext) { + parkNext = false; + await new Promise(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(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); + }); +});