test: add unit tests for SEP-1699 SSE retry field handling

felixweinberger · felixweinberger · commit 9ddf34d60015 · 2025-11-17T14:38:33.000Z
Client tests:
- Server-provided retry value used for reconnection delay
- Fallback to exponential backoff when no retry value
- Reconnection on graceful stream close with Last-Event-ID

Server tests:
- Send priming event with retry field when retryInterval configured
- No priming event when retryInterval not configured
- Event ID included in priming event for resumability
diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts
@@ -1010,6 +1010,148 @@ describe('StreamableHTTPClientTransport', () => {
         });
     });
 
+    describe('SSE retry field handling (SEP-1699)', () => {
+        beforeEach(() => {
+            vi.useFakeTimers();
+            (global.fetch as Mock).mockReset();
+        });
+        afterEach(() => vi.useRealTimers());
+
+        it('should use server-provided retry value for reconnection delay', async () => {
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                reconnectionOptions: {
+                    initialReconnectionDelay: 100,
+                    maxReconnectionDelay: 5000,
+                    reconnectionDelayGrowFactor: 2,
+                    maxRetries: 3
+                }
+            });
+
+            // Create a stream that sends a retry field
+            const encoder = new TextEncoder();
+            const stream = new ReadableStream({
+                start(controller) {
+                    // Send SSE event with retry field
+                    const event =
+                        'retry: 3000\nevent: message\nid: evt-1\ndata: {"jsonrpc": "2.0", "method": "notification", "params": {}}\n\n';
+                    controller.enqueue(encoder.encode(event));
+                    // Close stream to trigger reconnection
+                    controller.close();
+                }
+            });
+
+            const fetchMock = global.fetch as Mock;
+            fetchMock.mockResolvedValueOnce({
+                ok: true,
+                status: 200,
+                headers: new Headers({ 'content-type': 'text/event-stream' }),
+                body: stream
+            });
+
+            // Second request for reconnection
+            fetchMock.mockResolvedValueOnce({
+                ok: true,
+                status: 200,
+                headers: new Headers({ 'content-type': 'text/event-stream' }),
+                body: new ReadableStream()
+            });
+
+            await transport.start();
+            await transport['_startOrAuthSse']({});
+
+            // Wait for stream to close and reconnection to be scheduled
+            await vi.advanceTimersByTimeAsync(100);
+
+            // Verify the server retry value was captured
+            const transportInternal = transport as unknown as { _serverRetryMs?: number };
+            expect(transportInternal._serverRetryMs).toBe(3000);
+
+            // Verify the delay calculation uses server retry value
+            const getDelay = transport['_getNextReconnectionDelay'].bind(transport);
+            expect(getDelay(0)).toBe(3000); // Should use server value, not 100ms initial
+            expect(getDelay(5)).toBe(3000); // Should still use server value for any attempt
+        });
+
+        it('should fall back to exponential backoff when no server retry value', () => {
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                reconnectionOptions: {
+                    initialReconnectionDelay: 100,
+                    maxReconnectionDelay: 5000,
+                    reconnectionDelayGrowFactor: 2,
+                    maxRetries: 3
+                }
+            });
+
+            // Without any SSE stream, _serverRetryMs should be undefined
+            const transportInternal = transport as unknown as { _serverRetryMs?: number };
+            expect(transportInternal._serverRetryMs).toBeUndefined();
+
+            // Should use exponential backoff
+            const getDelay = transport['_getNextReconnectionDelay'].bind(transport);
+            expect(getDelay(0)).toBe(100); // 100 * 2^0
+            expect(getDelay(1)).toBe(200); // 100 * 2^1
+            expect(getDelay(2)).toBe(400); // 100 * 2^2
+            expect(getDelay(10)).toBe(5000); // capped at max
+        });
+
+        it('should reconnect on graceful stream close', async () => {
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                reconnectionOptions: {
+                    initialReconnectionDelay: 10,
+                    maxReconnectionDelay: 1000,
+                    reconnectionDelayGrowFactor: 1,
+                    maxRetries: 1
+                }
+            });
+
+            // Create a stream that closes gracefully after sending an event with ID
+            const encoder = new TextEncoder();
+            const stream = new ReadableStream({
+                start(controller) {
+                    // Send priming event with ID and retry field
+                    const event = 'id: evt-1\nretry: 100\ndata: \n\n';
+                    controller.enqueue(encoder.encode(event));
+                    // Graceful close
+                    controller.close();
+                }
+            });
+
+            const fetchMock = global.fetch as Mock;
+            fetchMock.mockResolvedValueOnce({
+                ok: true,
+                status: 200,
+                headers: new Headers({ 'content-type': 'text/event-stream' }),
+                body: stream
+            });
+
+            // Second request for reconnection
+            fetchMock.mockResolvedValueOnce({
+                ok: true,
+                status: 200,
+                headers: new Headers({ 'content-type': 'text/event-stream' }),
+                body: new ReadableStream()
+            });
+
+            await transport.start();
+            await transport['_startOrAuthSse']({});
+
+            // Wait for stream to process and close
+            await vi.advanceTimersByTimeAsync(50);
+
+            // Wait for reconnection delay (100ms from retry field)
+            await vi.advanceTimersByTimeAsync(150);
+
+            // Should have attempted reconnection
+            expect(fetchMock).toHaveBeenCalledTimes(2);
+            expect(fetchMock.mock.calls[0][1]?.method).toBe('GET');
+            expect(fetchMock.mock.calls[1][1]?.method).toBe('GET');
+
+            // Second call should include Last-Event-ID
+            const secondCallHeaders = fetchMock.mock.calls[1][1]?.headers;
+            expect(secondCallHeaders?.get('last-event-id')).toBe('evt-1');
+        });
+    });
+
     describe('prevent infinite recursion when server returns 401 after successful auth', () => {
         it('should throw error when server returns 401 after successful auth', async () => {
             const message: JSONRPCMessage = {
diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts
@@ -320,12 +320,14 @@ export class StreamableHTTPClientTransport implements Transport {
                 // Create a pipeline: binary stream -> text decoder -> SSE parser
                 const reader = stream
                     .pipeThrough(new TextDecoderStream() as ReadableWritablePair<string, Uint8Array>)
-                    .pipeThrough(new EventSourceParserStream({
-                        onRetry: (retryMs: number) => {
-                            // SEP-1699: Capture server-provided retry value for reconnection timing
-                            this._serverRetryMs = retryMs;
-                        }
-                    }))
+                    .pipeThrough(
+                        new EventSourceParserStream({
+                            onRetry: (retryMs: number) => {
+                                // SEP-1699: Capture server-provided retry value for reconnection timing
+                                this._serverRetryMs = retryMs;
+                            }
+                        })
+                    )
                     .getReader();
 
                 while (true) {
diff --git a/src/server/streamableHttp.test.ts b/src/server/streamableHttp.test.ts
@@ -31,6 +31,7 @@ interface TestServerConfig {
     eventStore?: EventStore;
     onsessioninitialized?: (sessionId: string) => void | Promise<void>;
     onsessionclosed?: (sessionId: string) => void | Promise<void>;
+    retryInterval?: number;
 }
 
 /**
@@ -58,7 +59,8 @@ async function createTestServer(config: TestServerConfig = { sessionIdGenerator:
         enableJsonResponse: config.enableJsonResponse ?? false,
         eventStore: config.eventStore,
         onsessioninitialized: config.onsessioninitialized,
-        onsessionclosed: config.onsessionclosed
+        onsessionclosed: config.onsessionclosed,
+        retryInterval: config.retryInterval
     });
 
     await mcpServer.connect(transport);
@@ -1516,6 +1518,141 @@ describe('StreamableHTTPServerTransport in stateless mode', () => {
     });
 });
 
+// Test SSE priming events (SEP-1699)
+describe('StreamableHTTPServerTransport SSE priming events (SEP-1699)', () => {
+    let server: Server;
+    let transport: StreamableHTTPServerTransport;
+    let baseUrl: URL;
+    let sessionId: string;
+
+    afterEach(async () => {
+        if (server && transport) {
+            await stopTestServer({ server, transport });
+        }
+    });
+
+    it('should send priming event with retry field when retryInterval is configured', async () => {
+        const result = await createTestServer({
+            sessionIdGenerator: () => randomUUID(),
+            retryInterval: 5000
+        });
+        server = result.server;
+        transport = result.transport;
+        baseUrl = result.baseUrl;
+
+        // Initialize to get session ID
+        const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
+        sessionId = initResponse.headers.get('mcp-session-id') as string;
+        expect(sessionId).toBeDefined();
+
+        // Open a standalone SSE stream
+        const sseResponse = await fetch(baseUrl, {
+            method: 'GET',
+            headers: {
+                Accept: 'text/event-stream',
+                'mcp-session-id': sessionId,
+                'mcp-protocol-version': '2025-03-26'
+            }
+        });
+
+        expect(sseResponse.status).toBe(200);
+        expect(sseResponse.headers.get('content-type')).toBe('text/event-stream');
+
+        // Read the priming event
+        const reader = sseResponse.body?.getReader();
+        const { value } = await reader!.read();
+        const text = new TextDecoder().decode(value);
+
+        // Verify priming event has id and retry field
+        expect(text).toContain('id: ');
+        expect(text).toContain('retry: 5000');
+        expect(text).toContain('data: ');
+    });
+
+    it('should not send priming event when retryInterval is not configured', async () => {
+        const result = await createTestServer({
+            sessionIdGenerator: () => randomUUID()
+            // No retryInterval
+        });
+        server = result.server;
+        transport = result.transport;
+        baseUrl = result.baseUrl;
+
+        // Initialize to get session ID
+        const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
+        sessionId = initResponse.headers.get('mcp-session-id') as string;
+        expect(sessionId).toBeDefined();
+
+        // Open a standalone SSE stream
+        const sseResponse = await fetch(baseUrl, {
+            method: 'GET',
+            headers: {
+                Accept: 'text/event-stream',
+                'mcp-session-id': sessionId,
+                'mcp-protocol-version': '2025-03-26'
+            }
+        });
+
+        expect(sseResponse.status).toBe(200);
+
+        // Send a notification to get some output
+        const notification: JSONRPCMessage = {
+            jsonrpc: '2.0',
+            method: 'notifications/message',
+            params: { level: 'info', data: 'Test' }
+        };
+        await transport.send(notification);
+
+        // Read from stream
+        const reader = sseResponse.body?.getReader();
+        const { value } = await reader!.read();
+        const text = new TextDecoder().decode(value);
+
+        // Should NOT have retry field (since retryInterval not configured)
+        expect(text).not.toContain('retry:');
+        // Should have the notification
+        expect(text).toContain('Test');
+    });
+
+    it('should include event ID in priming event for resumability', async () => {
+        const result = await createTestServer({
+            sessionIdGenerator: () => randomUUID(),
+            retryInterval: 2000
+        });
+        server = result.server;
+        transport = result.transport;
+        baseUrl = result.baseUrl;
+
+        // Initialize to get session ID
+        const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
+        sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        // Open SSE stream
+        const sseResponse = await fetch(baseUrl, {
+            method: 'GET',
+            headers: {
+                Accept: 'text/event-stream',
+                'mcp-session-id': sessionId,
+                'mcp-protocol-version': '2025-03-26'
+            }
+        });
+
+        // Read priming event
+        const reader = sseResponse.body?.getReader();
+        const { value } = await reader!.read();
+        const text = new TextDecoder().decode(value);
+
+        // Extract event ID from priming event
+        const idMatch = text.match(/id: ([^\n]+)/);
+        expect(idMatch).toBeTruthy();
+        const eventId = idMatch![1];
+
+        // Event ID should be present and non-empty
+        expect(eventId).toBeTruthy();
+        expect(eventId.length).toBeGreaterThan(0);
+    });
+});
+
 // Test onsessionclosed callback
 describe('StreamableHTTPServerTransport onsessionclosed callback', () => {
     it('should call onsessionclosed callback when session is closed via DELETE', async () => {
