fix: send priming events unconditionally for resumability (SEP-1699)

felixweinberger · felixweinberger · commit 54d359e926c6 · 2025-11-17T17:23:50.000Z
Per the spec, priming events (id + empty data) should always be sent
on SSE stream start for resumability. The retry field is separate and
should only be added when polling behavior is wanted via retryInterval.

Previously, priming events were only sent when retryInterval was set,
conflating resumability with polling. Now:
- Priming events are always sent on both GET and POST SSE streams
- Retry field is only included when retryInterval is configured
diff --git a/src/server/streamableHttp.test.ts b/src/server/streamableHttp.test.ts
@@ -184,8 +184,23 @@ const TEST_MESSAGES = {
  */
 async function readSSEEvent(response: Response): Promise<string> {
     const reader = response.body?.getReader();
-    const { value } = await reader!.read();
-    return new TextDecoder().decode(value);
+    // Keep reading until we get an event with actual data (skip priming events)
+    while (true) {
+        const { value, done } = await reader!.read();
+        if (done) break;
+        const text = new TextDecoder().decode(value);
+        // Check all data lines (multiple events may come in one read)
+        const dataLines = text.split('\n').filter(line => line.startsWith('data:'));
+        // Find first data line with actual content (not just empty priming event)
+        const dataLineWithContent = dataLines.find(line => {
+            const content = line.substring(5).trim();
+            return content !== '';
+        });
+        if (dataLineWithContent) {
+            return text;
+        }
+    }
+    return '';
 }
 
 /**
@@ -313,7 +328,7 @@ describe('StreamableHTTPServerTransport', () => {
 
         // Parse the SSE event
         const eventLines = text.split('\n');
-        const dataLine = eventLines.find(line => line.startsWith('data:'));
+        const dataLine = eventLines.find(line => line.startsWith('data:') && line.substring(5).trim() !== '');
         expect(dataLine).toBeDefined();
 
         const eventData = JSON.parse(dataLine!.substring(5));
@@ -351,7 +366,7 @@ describe('StreamableHTTPServerTransport', () => {
 
         const text = await readSSEEvent(response);
         const eventLines = text.split('\n');
-        const dataLine = eventLines.find(line => line.startsWith('data:'));
+        const dataLine = eventLines.find(line => line.startsWith('data:') && line.substring(5).trim() !== '');
         expect(dataLine).toBeDefined();
 
         const eventData = JSON.parse(dataLine!.substring(5));
@@ -406,7 +421,7 @@ describe('StreamableHTTPServerTransport', () => {
 
         const text = await readSSEEvent(response);
         const eventLines = text.split('\n');
-        const dataLine = eventLines.find(line => line.startsWith('data:'));
+        const dataLine = eventLines.find(line => line.startsWith('data:') && line.substring(5).trim() !== '');
         expect(dataLine).toBeDefined();
 
         const eventData = JSON.parse(dataLine!.substring(5));
@@ -489,7 +504,7 @@ describe('StreamableHTTPServerTransport', () => {
         const text = await readSSEEvent(sseResponse);
 
         const eventLines = text.split('\n');
-        const dataLine = eventLines.find(line => line.startsWith('data:'));
+        const dataLine = eventLines.find(line => line.startsWith('data:') && line.substring(5).trim() !== '');
         expect(dataLine).toBeDefined();
 
         const eventData = JSON.parse(dataLine!.substring(5));
@@ -516,6 +531,9 @@ describe('StreamableHTTPServerTransport', () => {
         expect(sseResponse.status).toBe(200);
         const reader = sseResponse.body?.getReader();
 
+        // Skip the priming event
+        await reader!.read();
+
         // Send multiple notifications
         const notification1: JSONRPCMessage = {
             jsonrpc: '2.0',
@@ -985,7 +1003,7 @@ describe('StreamableHTTPServerTransport with AuthInfo', () => {
 
         const text = await readSSEEvent(response);
         const eventLines = text.split('\n');
-        const dataLine = eventLines.find(line => line.startsWith('data:'));
+        const dataLine = eventLines.find(line => line.startsWith('data:') && line.substring(5).trim() !== '');
         expect(dataLine).toBeDefined();
 
         const eventData = JSON.parse(dataLine!.substring(5));
@@ -1021,7 +1039,7 @@ describe('StreamableHTTPServerTransport with AuthInfo', () => {
 
         const text = await readSSEEvent(response);
         const eventLines = text.split('\n');
-        const dataLine = eventLines.find(line => line.startsWith('data:'));
+        const dataLine = eventLines.find(line => line.startsWith('data:') && line.substring(5).trim() !== '');
         expect(dataLine).toBeDefined();
 
         const eventData = JSON.parse(dataLine!.substring(5));
@@ -1352,6 +1370,10 @@ describe('StreamableHTTPServerTransport with resumability', () => {
 
         // Read from the stream and verify we got the notification with an event ID
         const reader = sseResponse.body?.getReader();
+
+        // Skip the priming event
+        await reader!.read();
+
         const { value } = await reader!.read();
         const text = new TextDecoder().decode(value);
 
@@ -1388,6 +1410,10 @@ describe('StreamableHTTPServerTransport with resumability', () => {
 
         // Read the notification from the SSE stream
         const reader = sseResponse.body?.getReader();
+
+        // Skip the priming event
+        await reader!.read();
+
         const { value } = await reader!.read();
         const text = new TextDecoder().decode(value);
 
@@ -1569,7 +1595,7 @@ describe('StreamableHTTPServerTransport SSE priming events (SEP-1699)', () => {
         expect(text).toContain('data: ');
     });
 
-    it('should not send priming event when retryInterval is not configured', async () => {
+    it('should send priming event without retry field when retryInterval is not configured', async () => {
         const result = await createTestServer({
             sessionIdGenerator: () => randomUUID()
             // No retryInterval
@@ -1595,23 +1621,15 @@ describe('StreamableHTTPServerTransport SSE priming events (SEP-1699)', () => {
 
         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
+        // Read the priming event
         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)
+        // Should have id field for resumability but NOT retry field
+        expect(text).toContain('id: ');
+        expect(text).toContain('data: ');
         expect(text).not.toContain('retry:');
-        // Should have the notification
-        expect(text).toContain('Test');
     });
 
     it('should include event ID in priming event for resumability', async () => {
diff --git a/src/server/streamableHttp.ts b/src/server/streamableHttp.ts
@@ -330,16 +330,17 @@ export class StreamableHTTPServerTransport implements Transport {
         res.writeHead(200, headers).flushHeaders();
 
         // SEP-1699: Send priming event with id and empty data to establish resumption capability
-        // This primes the client's Last-Event-ID for reconnection and sets retry interval
-        // Only send when retryInterval is configured (enabling polling behavior)
-        if (this._retryInterval !== undefined) {
-            const primingEventId = this._eventStore
-                ? await this._eventStore.storeEvent(this._standaloneSseStreamId, {} as JSONRPCMessage)
-                : `priming-${Date.now()}`;
+        // This primes the client's Last-Event-ID for reconnection
+        const primingEventId = this._eventStore
+            ? await this._eventStore.storeEvent(this._standaloneSseStreamId, {} as JSONRPCMessage)
+            : `priming-${Date.now()}`;
 
-            const primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
-            res.write(primingEvent);
+        // Build priming event - always include id for resumability, optionally add retry for polling
+        let primingEvent = `id: ${primingEventId}\ndata: \n\n`;
+        if (this._retryInterval !== undefined) {
+            primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
         }
+        res.write(primingEvent);
 
         // Assign the response to the standalone SSE stream
         this._streamMapping.set(this._standaloneSseStreamId, res);
@@ -570,16 +571,17 @@ export class StreamableHTTPServerTransport implements Transport {
                     res.writeHead(200, headers);
 
                     // SEP-1699: Send priming event with id and empty data to establish resumption capability
-                    // This primes the client's Last-Event-ID for reconnection and sets retry interval
-                    // Only send when retryInterval is configured (enabling polling behavior)
-                    if (this._retryInterval !== undefined) {
-                        const primingEventId = this._eventStore
-                            ? await this._eventStore.storeEvent(streamId, {} as JSONRPCMessage)
-                            : `priming-${Date.now()}`;
+                    // This primes the client's Last-Event-ID for reconnection
+                    const primingEventId = this._eventStore
+                        ? await this._eventStore.storeEvent(streamId, {} as JSONRPCMessage)
+                        : `priming-${Date.now()}`;
 
-                        const primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
-                        res.write(primingEvent);
+                    // Build priming event - always include id for resumability, optionally add retry for polling
+                    let primingEvent = `id: ${primingEventId}\ndata: \n\n`;
+                    if (this._retryInterval !== undefined) {
+                        primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
                     }
+                    res.write(primingEvent);
                 }
                 // Store the response for this request to send messages back through this connection
                 // We need to track by request ID to maintain the connection
