Fix SSE JSON payload and empty stream offset bugs

Fixes identified during code review:

1. JSON SSE payload trailing comma bug:
   - processJsonAppend adds trailing commas for storage
   - SSE was wrapping raw data in [], producing invalid JSON like [{"id":1},]
   - Now uses formatResponse() which properly strips trailing commas

2. Empty stream offset bug:
   - SSE control events were using initialOffset (-1) for empty streams
   - HTTP GET uses stream.currentOffset for empty results
   - SSE now computes controlOffset the same way as HTTP GET

3. Strengthened conformance tests:
   - JSON SSE test now parses payload and verifies it's valid JSON
   - Empty stream test verifies SSE offset matches HTTP GET offset

Author: claude

packages/conformance-tests/src/index.ts

@@ -2457,7 +2457,7 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
       }
     })
 
-    test(`should wrap JSON data in arrays for SSE`, async () => {
+    test(`should wrap JSON data in arrays for SSE and produce valid JSON`, async () => {
       const streamPath = `/v1/stream/sse-json-wrap-test-${Date.now()}`
 
       // Create JSON stream
@@ -2499,8 +2499,19 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
 
         // JSON data should be wrapped in array brackets
         expect(received).toContain(`event: data`)
-        // The data line should contain the JSON wrapped in []
-        expect(received).toMatch(/data:.*\[.*\{.*"id".*:.*1.*\}.*\]/)
+
+        // 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)
+
+        // 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`) {
@@ -2509,7 +2520,7 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
       }
     })
 
-    test(`should handle SSE for empty stream`, async () => {
+    test(`should handle SSE for empty stream with correct offset`, async () => {
       const streamPath = `/v1/stream/sse-empty-test-${Date.now()}`
 
       // Create empty stream
@@ -2518,6 +2529,12 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
         headers: { "Content-Type": `text/plain` },
       })
 
+      // First, get the offset from HTTP GET (the canonical source)
+      const httpResponse = await fetch(`${getBaseUrl()}${streamPath}`)
+      const httpOffset = httpResponse.headers.get(`Stream-Next-Offset`)
+      expect(httpOffset).toBeDefined()
+      expect(httpOffset).not.toBe(`-1`) // Should be the stream's actual offset, not -1
+
       // Make SSE request
       const controller = new AbortController()
       const timeoutId = setTimeout(() => controller.abort(), 1000)
@@ -2550,6 +2567,20 @@ export function runConformanceTests(options: ConformanceTestOptions): void {
 
         // Should get a control event even for empty stream
         expect(received).toContain(`event: control`)
+
+        // Parse the control event and verify offset matches HTTP GET
+        const controlLine = received
+          .split(`\n`)
+          .find(
+            (l) => l.startsWith(`data: `) && l.includes(`Stream-Next-Offset`)
+          )
+        expect(controlLine).toBeDefined()
+
+        const controlPayload = controlLine!.slice(`data: `.length)
+        const controlData = JSON.parse(controlPayload)
+
+        // SSE control offset should match HTTP GET offset (not -1)
+        expect(controlData[`Stream-Next-Offset`]).toBe(httpOffset)
       } catch (e) {
         clearTimeout(timeoutId)
         if (e instanceof Error && e.name !== `AbortError`) {

packages/server/src/server.ts

@@ -515,6 +515,7 @@ export class DurableStreamTestServer {
 
     let currentOffset = initialOffset
     let isConnected = true
+    const decoder = new TextDecoder()
 
     // Handle client disconnect
     res.on(`close`, () => {
@@ -537,10 +538,11 @@ export class DurableStreamTestServer {
         // Format data based on content type
         let dataPayload: string
         if (isJsonStream) {
-          // Wrap JSON in array brackets for SSE
-          dataPayload = `[${new TextDecoder().decode(message.data)}]`
+          // Use formatResponse to get properly formatted JSON (strips trailing commas)
+          const jsonBytes = this.store.formatResponse(path, [message])
+          dataPayload = decoder.decode(jsonBytes)
         } else {
-          dataPayload = new TextDecoder().decode(message.data)
+          dataPayload = decoder.decode(message.data)
         }
 
         // Send data event
@@ -550,9 +552,13 @@ export class DurableStreamTestServer {
         currentOffset = message.offset
       }
 
+      // Compute offset the same way as HTTP GET: last message's offset, or stream's current offset
+      const controlOffset =
+        messages[messages.length - 1]?.offset ?? stream!.currentOffset
+
       // Send control event with current offset/cursor
       const controlData: Record<string, string> = {
-        [STREAM_OFFSET_HEADER]: currentOffset,
+        [STREAM_OFFSET_HEADER]: controlOffset,
       }
       if (cursor) {
         controlData[STREAM_CURSOR_HEADER] = cursor
@@ -561,6 +567,9 @@ export class DurableStreamTestServer {
       res.write(`event: control\n`)
       res.write(`data: ${JSON.stringify(controlData)}\n\n`)
 
+      // Update currentOffset for next iteration (use controlOffset for consistency)
+      currentOffset = controlOffset
+
       // If caught up, wait for new messages
       if (upToDate) {
         const result = await this.store.waitForMessages(