Fix SSE multiline encoding and add test helper

Reviewer feedback fixes:

1. SSE multiline payload encoding:
   - Per SSE spec, each line in payload needs its own "data:" prefix
   - Added encodeSSEData() helper that splits on newlines
   - "line1\nline2" now becomes "data: line1\ndata: line2\n\n"

2. Test helper for SSE fetch pattern:
   - Added fetchSSE() helper to reduce boilerplate
   - Handles AbortController, timeout, chunked reading, cleanup
   - Refactored cursor, JSON, and newline tests to use helper

3. Strengthened newline test:
   - Now verifies each line has its own "data:" prefix
   - Catches SSE spec violations

Author: claude

packages/conformance-tests/src/index.ts

@@ -18,6 +18,61 @@ export interface ConformanceTestOptions {
   baseUrl: string
 }
 
+/**
+ * Helper to fetch SSE stream and read until a condition is met.
+ * Handles AbortController, timeout, and cleanup automatically.
+ */
+async function fetchSSE(
+  url: string,
+  opts: {
+    timeoutMs?: number
+    maxChunks?: number
+    untilContent?: string
+    headers?: Record<string, string>
+  } = {}
+): Promise<{ response: Response; received: string }> {
+  const { timeoutMs = 2000, maxChunks = 10, untilContent, headers = {} } = opts
+
+  const controller = new AbortController()
+  const timeoutId = setTimeout(() => controller.abort(), timeoutMs)
+
+  try {
+    const response = await fetch(url, {
+      method: `GET`,
+      headers,
+      signal: controller.signal,
+    })
+
+    if (!response.body) {
+      clearTimeout(timeoutId)
+      return { response, received: `` }
+    }
+
+    const reader = response.body.getReader()
+    const decoder = new TextDecoder()
+    let received = ``
+
+    for (let i = 0; i < maxChunks; i++) {
+      const { done, value } = await reader.read()
+      if (done) break
+      received += decoder.decode(value, { stream: true })
+      if (untilContent && received.includes(untilContent)) break
+    }
+
+    clearTimeout(timeoutId)
+    reader.cancel()
+
+    return { response, received }
+  } catch (e) {
+    clearTimeout(timeoutId)
+    if (e instanceof Error && e.name === `AbortError`) {
+      // Return empty result on timeout/abort
+      return { response: new Response(), received: `` }
+    }
+    throw e
+  }
+}
+
 /**
  * Run the full conformance test suite against a server
  */
@@ -2418,43 +2473,14 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
         body: `test data`,
       })
 
-      // SSE request with cursor should be accepted
-      const controller = new AbortController()
-      const timeoutId = setTimeout(() => controller.abort(), 1000)
-
-      try {
-        const response = await fetch(
-          `${getBaseUrl()}${streamPath}?offset=-1&live=sse&cursor=test-cursor-456`,
-          {
-            method: `GET`,
-          }
-        )
-
-        clearTimeout(timeoutId)
-        expect(response.status).toBe(200)
-
-        // Read a bit and verify cursor is echoed in control event
-        const reader = response.body!.getReader()
-        const decoder = new TextDecoder()
-        let received = ``
-
-        for (let i = 0; i < 10; i++) {
-          const { done, value } = await reader.read()
-          if (done) break
-          received += decoder.decode(value, { stream: true })
-          if (received.includes(`Stream-Cursor`)) break
-        }
-
-        reader.cancel()
+      const { response, received } = await fetchSSE(
+        `${getBaseUrl()}${streamPath}?offset=-1&live=sse&cursor=test-cursor-456`,
+        { untilContent: `Stream-Cursor` }
+      )
 
-        // Cursor should be echoed in control events
-        expect(received).toContain(`test-cursor-456`)
-      } catch (e) {
-        clearTimeout(timeoutId)
-        if (e instanceof Error && e.name !== `AbortError`) {
-          throw e
-        }
-      }
+      expect(response.status).toBe(200)
+      // Cursor should be echoed in control events
+      expect(received).toContain(`test-cursor-456`)
     })
 
     test(`should wrap JSON data in arrays for SSE and produce valid JSON`, async () => {
@@ -2467,57 +2493,26 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
         body: JSON.stringify({ id: 1, message: `hello` }),
       })
 
-      // Make SSE request
-      const controller = new AbortController()
-      const timeoutId = setTimeout(() => controller.abort(), 2000)
-
-      try {
-        const response = await fetch(
-          `${getBaseUrl()}${streamPath}?offset=-1&live=sse`,
-          {
-            method: `GET`,
-            signal: controller.signal,
-          }
-        )
-
-        expect(response.status).toBe(200)
-
-        // Read response
-        const reader = response.body!.getReader()
-        const decoder = new TextDecoder()
-        let received = ``
-
-        for (let i = 0; i < 10; i++) {
-          const { done, value } = await reader.read()
-          if (done) break
-          received += decoder.decode(value, { stream: true })
-          if (received.includes(`event: data`)) break
-        }
-
-        clearTimeout(timeoutId)
-        reader.cancel()
+      const { response, received } = await fetchSSE(
+        `${getBaseUrl()}${streamPath}?offset=-1&live=sse`,
+        { untilContent: `event: data` }
+      )
 
-        // JSON data should be wrapped in array brackets
-        expect(received).toContain(`event: data`)
+      expect(response.status).toBe(200)
+      expect(received).toContain(`event: data`)
 
-        // Extract and parse the JSON payload to verify it's valid
-        const dataLine = received
-          .split(`\n`)
-          .find((l) => l.startsWith(`data: `) && l.includes(`[`))
-        expect(dataLine).toBeDefined()
+      // Extract and parse the JSON payload to verify it's valid
+      const dataLine = received
+        .split(`\n`)
+        .find((l) => l.startsWith(`data: `) && l.includes(`[`))
+      expect(dataLine).toBeDefined()
 
-        const payload = dataLine!.slice(`data: `.length)
-        // This will throw if JSON is invalid (e.g., trailing comma)
-        const parsed = JSON.parse(payload)
+      const payload = dataLine!.slice(`data: `.length)
+      // This will throw if JSON is invalid (e.g., trailing comma)
+      const parsed = JSON.parse(payload)
 
-        // Verify the structure matches what we sent
-        expect(parsed).toEqual([{ id: 1, message: `hello` }])
-      } catch (e) {
-        clearTimeout(timeoutId)
-        if (e instanceof Error && e.name !== `AbortError`) {
-          throw e
-        }
-      }
+      // Verify the structure matches what we sent
+      expect(parsed).toEqual([{ id: 1, message: `hello` }])
     })
 
     test(`should handle SSE for empty stream with correct offset`, async () => {
@@ -2643,53 +2638,19 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
         body: `line1\nline2\nline3`,
       })
 
-      // Make SSE request
-      const controller = new AbortController()
-      const timeoutId = setTimeout(() => controller.abort(), 2000)
-
-      try {
-        const response = await fetch(
-          `${getBaseUrl()}${streamPath}?offset=-1&live=sse`,
-          {
-            method: `GET`,
-            signal: controller.signal,
-          }
-        )
-
-        expect(response.status).toBe(200)
-
-        // Read response
-        const reader = response.body!.getReader()
-        const decoder = new TextDecoder()
-        let received = ``
-
-        for (let i = 0; i < 10; i++) {
-          const { done, value } = await reader.read()
-          if (done) break
-          received += decoder.decode(value, { stream: true })
-          if (received.includes(`event: control`)) break
-        }
+      const { response, received } = await fetchSSE(
+        `${getBaseUrl()}${streamPath}?offset=-1&live=sse`,
+        { untilContent: `event: control` }
+      )
 
-        clearTimeout(timeoutId)
-        reader.cancel()
+      expect(response.status).toBe(200)
+      expect(received).toContain(`event: data`)
 
-        // The SSE data field should contain the text
-        // Note: SSE spec says newlines in data need multiple data: lines
-        // or the newlines become field separators
-        expect(received).toContain(`event: data`)
-
-        // At minimum, the original content should be recoverable
-        // (server may encode newlines as multiple data: lines per SSE spec)
-        const hasLine1 = received.includes(`line1`)
-        const hasLine2 = received.includes(`line2`)
-        const hasLine3 = received.includes(`line3`)
-        expect(hasLine1 && hasLine2 && hasLine3).toBe(true)
-      } catch (e) {
-        clearTimeout(timeoutId)
-        if (e instanceof Error && e.name !== `AbortError`) {
-          throw e
-        }
-      }
+      // Per SSE spec, multiline data must use multiple "data:" lines
+      // Each line should have its own data: prefix
+      expect(received).toContain(`data: line1`)
+      expect(received).toContain(`data: line2`)
+      expect(received).toContain(`data: line3`)
     })
 
     test(`should maintain monotonic offsets over multiple messages`, async () => {

packages/server/src/server.ts

@@ -21,6 +21,16 @@ const OFFSET_QUERY_PARAM = `offset`
 const LIVE_QUERY_PARAM = `live`
 const CURSOR_QUERY_PARAM = `cursor`
 
+/**
+ * Encode data for SSE format.
+ * Per SSE spec, each line in the payload needs its own "data:" prefix.
+ * Newlines in the payload become separate data: lines.
+ */
+function encodeSSEData(payload: string): string {
+  const lines = payload.split(`\n`)
+  return lines.map((line) => `data: ${line}`).join(`\n`) + `\n\n`
+}
+
 /**
  * HTTP server for testing durable streams.
  * Supports both in-memory and file-backed storage modes.
@@ -545,9 +555,10 @@ export class DurableStreamTestServer {
           dataPayload = decoder.decode(message.data)
         }
 
-        // Send data event
+        // Send data event - encode multiline payloads per SSE spec
+        // Each line in the payload needs its own "data:" prefix
         res.write(`event: data\n`)
-        res.write(`data: ${dataPayload}\n\n`)
+        res.write(encodeSSEData(dataPayload))
 
         currentOffset = message.offset
       }
@@ -565,7 +576,7 @@ export class DurableStreamTestServer {
       }
 
       res.write(`event: control\n`)
-      res.write(`data: ${JSON.stringify(controlData)}\n\n`)
+      res.write(encodeSSEData(JSON.stringify(controlData)))
 
       // Update currentOffset for next iteration (use controlOffset for consistency)
       currentOffset = controlOffset
@@ -591,7 +602,7 @@ export class DurableStreamTestServer {
             keepAliveData[STREAM_CURSOR_HEADER] = cursor
           }
           res.write(`event: control\n`)
-          res.write(`data: ${JSON.stringify(keepAliveData)}\n\n`)
+          res.write(encodeSSEData(JSON.stringify(keepAliveData)))
         }
         // Loop will continue and read new messages
       }