fix(server): streamableHttp stores request-related events when stream is disconnected

The standalone-SSE path stores to eventStore first, then writes if connected.
The request-related path only stored when the stream was live, so a
notification sent after closeSSE() (SEP-1699 polling) was silently dropped
instead of being persisted for replay on reconnect.

Exposed by the ctx.mcpReq.log request-related change against the new
sse-polling example story; the gap pre-exists on main for any
request-related notification (progress, ctx.mcpReq.notify) emitted after
closeSSE().

Author: felixweinberger

.changeset/server-streamablehttp-store-first.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/server': patch
+---
+
+`WebStandardStreamableHTTPServerTransport`: request-related events (progress, `ctx.mcpReq.notify`, handler-emitted log) and the final response sent while the per-request SSE stream is intentionally closed (e.g. after `closeSSE()` for SEP-1699 polling) are now persisted to the configured `eventStore` so they replay on reconnect, and the final response no longer surfaces a spurious "No connection established" send error. Previously they were silently dropped. Events sent after a hard client disconnect (the per-request stream cancelled by the client) are not persisted.

packages/server/src/server/streamableHttp.ts

@@ -941,9 +941,24 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         if (!streamId) return;
 
         const stream = this._streamMapping.get(streamId);
-        if (stream) {
-            stream.cleanup();
+        if (stream?.controller) {
+            try {
+                stream.controller.close();
+            } catch {
+                // Controller might already be closed
+            }
         }
+        // Keep a placeholder _streamMapping entry (controller/encoder cleared)
+        // so send() can distinguish an intentional poll-and-replay close from a
+        // hard client disconnect (the ReadableStream cancel callback deletes
+        // the entry entirely). The placeholder is removed when the final
+        // response is sent (or overwritten when the client reconnects via
+        // Last-Event-ID and replayEvents() registers the resumed stream).
+        this._streamMapping.set(streamId, {
+            cleanup: () => {
+                this._streamMapping.delete(streamId);
+            }
+        });
     }
 
     /**
@@ -1002,15 +1017,24 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
 
         const 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 FIRST so request-related events emitted while the per-request
+            // stream is disconnected (e.g. after `closeSSE()`) are replayed on
+            // reconnect — same store-first semantics as the standalone path above.
+            // Gated on `stream !== undefined`: `closeSSEStream()` keeps a
+            // placeholder entry (controller/encoder cleared), so the intentional
+            // poll-and-replay case stores; a hard client disconnect (the
+            // ReadableStream cancel callback) deletes the entry entirely, so an
+            // abandoned request does not persist for a replay that will never be
+            // requested.
             let eventId: string | undefined;
-
-            if (this._eventStore) {
+            if (this._eventStore && stream !== undefined) {
                 eventId = await this._eventStore.storeEvent(streamId, message);
             }
-            // Write the event to the response stream
-            this.writeSSEEvent(stream.controller, stream.encoder, message, eventId);
+            if (stream?.controller && stream?.encoder) {
+                // Write the event to the response stream
+                this.writeSSEEvent(stream.controller, stream.encoder, message, eventId);
+            }
         }
 
         if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {

packages/server/test/server/streamableHttp.test.ts

@@ -705,6 +705,104 @@ describe('Zod v4', () => {
             // Should have id: field in the SSE event
             expect(text).toContain('id:');
         });
+
+        it('should store request-related events emitted after closeSSEStream() and not throw on the final response', async () => {
+            // The SEP-1699 poll-and-replay flow: handler closes the per-request
+            // SSE stream, then emits a notification and its final result while
+            // the client has not yet reconnected. Both must be persisted to the
+            // eventStore (so they replay on Last-Event-ID reconnect) and the
+            // final-response send must not surface a spurious error.
+            mcpServer.registerTool(
+                'poll',
+                { description: 'closeSSE then emit', inputSchema: z.object({}) },
+                async (_args, ctx): Promise<CallToolResult> => {
+                    ctx.http?.closeSSE?.();
+                    await ctx.mcpReq.notify({
+                        method: 'notifications/progress',
+                        params: { progressToken: 'poll-1', progress: 75 }
+                    });
+                    return { content: [{ type: 'text', text: 'done' }] };
+                }
+            );
+
+            const sendErrors: unknown[] = [];
+            mcpServer.server.onerror = e => sendErrors.push(e);
+
+            sessionId = await initializeServer();
+            storedEvents.clear();
+
+            const callMessage: JSONRPCMessage = {
+                jsonrpc: '2.0',
+                method: 'tools/call',
+                params: { name: 'poll', arguments: {} },
+                id: 'poll-1'
+            };
+            const response = await transport.handleRequest(createRequest('POST', callMessage, { sessionId }));
+            // closeSSE() in the handler closes the controller; drain the (now
+            // closed) body so the Response is fully consumed.
+            await response.text().catch(() => {});
+            // Let the async handler chain (notify + final response send) settle.
+            await new Promise(resolve => setTimeout(resolve, 50));
+
+            const stored = [...storedEvents.values()].map(e => e.message);
+            expect(
+                stored.some(m => 'method' in m && m.method === 'notifications/progress'),
+                'progress notification should be stored for replay after closeSSE()'
+            ).toBe(true);
+            expect(
+                stored.some(m => 'id' in m && m.id === 'poll-1' && 'result' in m),
+                'final response should be stored for replay after closeSSE()'
+            ).toBe(true);
+            expect(sendErrors).toEqual([]);
+        });
+
+        it('should not store request-related events after a hard client disconnect (stream cancel)', async () => {
+            // A hard client disconnect (the per-request ReadableStream's cancel
+            // callback) deletes the _streamMapping entry entirely; events emitted
+            // after that are NOT persisted (the request was abandoned).
+            let release!: () => void;
+            const gate = new Promise<void>(resolve => {
+                release = resolve;
+            });
+            mcpServer.registerTool(
+                'abandon',
+                { description: 'emit after client disconnect', inputSchema: z.object({}) },
+                async (_args, ctx): Promise<CallToolResult> => {
+                    await gate;
+                    await ctx.mcpReq.notify({
+                        method: 'notifications/progress',
+                        params: { progressToken: 'abandon-1', progress: 50 }
+                    });
+                    return { content: [{ type: 'text', text: 'done' }] };
+                }
+            );
+            // The final-response send still throws on a hard disconnect (pre-existing);
+            // swallow it so the test only pins the storage behavior.
+            mcpServer.server.onerror = () => {};
+
+            sessionId = await initializeServer();
+
+            const callMessage: JSONRPCMessage = {
+                jsonrpc: '2.0',
+                method: 'tools/call',
+                params: { name: 'abandon', arguments: {} },
+                id: 'abandon-1'
+            };
+            const response = await transport.handleRequest(createRequest('POST', callMessage, { sessionId }));
+            storedEvents.clear();
+            // Hard-disconnect: cancel the per-request stream (fires the
+            // ReadableStream cancel callback, deleting the _streamMapping entry).
+            await response.body?.cancel();
+            await new Promise(resolve => setTimeout(resolve, 10));
+            release();
+            await new Promise(resolve => setTimeout(resolve, 50));
+
+            const stored = [...storedEvents.values()].map(e => e.message);
+            expect(
+                stored.some(m => 'method' in m && m.method === 'notifications/progress'),
+                'progress notification should NOT be stored after hard client disconnect'
+            ).toBe(false);
+        });
     });
 
     describe('HTTPServerTransport - Protocol Version Validation', () => {