fix: only send priming events when eventStore is configured

Priming events are only meaningful for resumability, which requires an
eventStore. Without eventStore, the event ID is useless since there's
nothing to resume from.

- Rename _writePrimingEvent to _maybeWritePrimingEvent
- Guard priming event behind eventStore check
- Restore original tests that don't have eventStore
- Add eventStore to priming event tests

Author: felixweinberger

src/server/streamableHttp.test.ts

@@ -534,20 +534,14 @@ describe('StreamableHTTPServerTransport', () => {
         expect(sseResponse.status).toBe(200);
         const reader = sseResponse.body?.getReader();
 
-        // Read and validate the priming event
-        const { value: primingValue } = await reader!.read();
-        const primingText = new TextDecoder().decode(primingValue);
-        expect(primingText).toContain('id: ');
-        expect(primingText.match(/id: ([^\n]+)/)).toBeTruthy();
-
-        // Send a notification
+        // Send multiple notifications
         const notification1: JSONRPCMessage = {
             jsonrpc: '2.0',
             method: 'notifications/message',
             params: { level: 'info', data: 'First notification' }
         };
 
-        // Send and verify it comes through - then the stream should stay open
+        // Just send one and verify it comes through - then the stream should stay open
         await transport.send(notification1);
 
         const { value, done } = await reader!.read();
@@ -1562,6 +1556,30 @@ describe('StreamableHTTPServerTransport SSE priming events', () => {
     let baseUrl: URL;
     let sessionId: string;
 
+    // Simple eventStore for priming event tests
+    const createEventStore = (): EventStore => {
+        const storedEvents = new Map<string, { eventId: string; message: JSONRPCMessage }>();
+        return {
+            async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
+                const eventId = `${streamId}_${randomUUID()}`;
+                storedEvents.set(eventId, { eventId, message });
+                return eventId;
+            },
+            async replayEventsAfter(
+                lastEventId: EventId,
+                { send }: { send: (eventId: EventId, message: JSONRPCMessage) => Promise<void> }
+            ): Promise<StreamId> {
+                const streamId = lastEventId.split('_')[0];
+                for (const [eventId, { message }] of storedEvents.entries()) {
+                    if (eventId.startsWith(streamId) && eventId !== lastEventId) {
+                        await send(eventId, message);
+                    }
+                }
+                return streamId;
+            }
+        };
+    };
+
     afterEach(async () => {
         if (server && transport) {
             await stopTestServer({ server, transport });
@@ -1571,6 +1589,7 @@ describe('StreamableHTTPServerTransport SSE priming events', () => {
     it('should send priming event with retry field when retryInterval is configured', async () => {
         const result = await createTestServer({
             sessionIdGenerator: () => randomUUID(),
+            eventStore: createEventStore(),
             retryInterval: 5000
         });
         server = result.server;
@@ -1608,7 +1627,8 @@ describe('StreamableHTTPServerTransport SSE priming events', () => {
 
     it('should send priming event without retry field when retryInterval is not configured', async () => {
         const result = await createTestServer({
-            sessionIdGenerator: () => randomUUID()
+            sessionIdGenerator: () => randomUUID(),
+            eventStore: createEventStore()
             // No retryInterval
         });
         server = result.server;
@@ -1665,6 +1685,7 @@ describe('StreamableHTTPServerTransport SSE priming events', () => {
     it('should include event ID in priming event for resumability', async () => {
         const result = await createTestServer({
             sessionIdGenerator: () => randomUUID(),
+            eventStore: createEventStore(),
             retryInterval: 2000
         });
         server = result.server;

src/server/streamableHttp.ts

@@ -260,10 +260,14 @@ export class StreamableHTTPServerTransport implements Transport {
 
     /**
      * Writes a priming event to establish resumption capability.
-     * This primes the client's Last-Event-ID for reconnection.
+     * Only sends if eventStore is configured (opt-in for resumability).
      */
-    private async _writePrimingEvent(res: ServerResponse, streamId: string): Promise<void> {
-        const primingEventId = this._eventStore ? await this._eventStore.storeEvent(streamId, {} as JSONRPCMessage) : randomUUID();
+    private async _maybeWritePrimingEvent(res: ServerResponse, streamId: string): Promise<void> {
+        if (!this._eventStore) {
+            return;
+        }
+
+        const primingEventId = await this._eventStore.storeEvent(streamId, {} as JSONRPCMessage);
 
         let primingEvent = `id: ${primingEventId}\ndata: \n\n`;
         if (this._retryInterval !== undefined) {
@@ -343,7 +347,7 @@ export class StreamableHTTPServerTransport implements Transport {
         // otherwise the client will just wait for the first message
         res.writeHead(200, headers).flushHeaders();
 
-        await this._writePrimingEvent(res, this._standaloneSseStreamId);
+        await this._maybeWritePrimingEvent(res, this._standaloneSseStreamId);
 
         // Assign the response to the standalone SSE stream
         this._streamMapping.set(this._standaloneSseStreamId, res);
@@ -573,7 +577,7 @@ export class StreamableHTTPServerTransport implements Transport {
 
                     res.writeHead(200, headers);
 
-                    await this._writePrimingEvent(res, streamId);
+                    await this._maybeWritePrimingEvent(res, streamId);
                 }
                 // Store the response for this request to send messages back through this connection
                 // We need to track by request ID to maintain the connection