fix(stream): handle SSE line endings and disable proxy buffering (#636)

mateclaw-server/src/main/java/vip/mate/channel/web/Utf8SseEmitter.java

@@ -45,6 +45,10 @@ public class Utf8SseEmitter extends SseEmitter {
     protected void extendResponse(ServerHttpResponse response) {
         super.extendResponse(response);
         HttpHeaders headers = response.getHeaders();
+        // Streaming frames must reach the client as they are emitted. Nginx
+        // honors this response header unless explicitly configured to ignore it.
+        headers.set("X-Accel-Buffering", "no");
+        headers.setCacheControl("no-store, no-transform");
         // Spring's default sets Content-Type=text/event-stream without charset.
         // Only override when no charset is already specified, so callers that
         // want to roll their own (rare) keep working.

mateclaw-server/src/test/java/vip/mate/channel/web/Utf8SseEmitterTest.java

@@ -22,6 +22,16 @@ import static org.junit.jupiter.api.Assertions.*;
  */
 class Utf8SseEmitterTest {
 
+    @Test
+    void disablesProxyBufferingAndCachingWhileKeepingUtf8() {
+        var response = new ServletServerHttpResponse(new MockHttpServletResponse());
+        new Utf8SseEmitter().extendResponse(response);
+
+        assertEquals("no", response.getHeaders().getFirst("X-Accel-Buffering"));
+        assertEquals("no-store, no-transform", response.getHeaders().getCacheControl());
+        assertEquals(StandardCharsets.UTF_8, response.getHeaders().getContentType().getCharset());
+    }
+
     @Test
     @DisplayName("extendResponse stamps charset=UTF-8 when Content-Type is unset")
     void stampsUtf8WhenContentTypeUnset() throws Exception {

mateclaw-ui/src/composables/chat/__tests__/useStreamFraming.test.ts

@@ -0,0 +1,59 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { useStream } from '../useStream'
+
+afterEach(() => vi.unstubAllGlobals())
+
+describe('incremental SSE framing', () => {
+  it.each(['\n', '\r\n', '\r'])('dispatches %j frames before the connection closes', async (newline) => {
+    let controller!: ReadableStreamDefaultController<Uint8Array>
+    const body = new ReadableStream<Uint8Array>({ start(c) { controller = c } })
+    vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(body)))
+    const stream = useStream({ url: '/test' })
+    const received: string[] = []
+    stream.on('content_delta', data => received.push(data.delta))
+    const connected = stream.connect()
+    const encoder = new TextEncoder()
+    try {
+      for (const delta of ['first', 'second', 'third']) {
+        const frame = `event: content_delta${newline}data: ${JSON.stringify({ delta })}${newline}${newline}`
+        // Every possible CRLF pair is split across network reads.
+        for (const char of frame) controller.enqueue(encoder.encode(char))
+        await vi.waitFor(() => expect(received.at(-1)).toBe(delta), { timeout: 300 })
+      }
+      expect(received).toEqual(['first', 'second', 'third'])
+    } finally {
+      controller.close()
+      await connected
+    }
+  })
+
+  it('joins multiple data lines and ignores comment-only frames', async () => {
+    const payload = ': heartbeat\r\n\r\nevent: content_delta\r\ndata: {"delta":\r\ndata: "你好"}\r\n\r\n'
+    vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(payload)))
+    const stream = useStream({ url: '/test' })
+    const received: unknown[] = []
+    stream.onEvent(event => received.push(event.data))
+    await stream.connect()
+    expect(received).toEqual([{ delta: '你好' }])
+  })
+
+  it('keeps streaming when line endings change after the first event', async () => {
+    let controller!: ReadableStreamDefaultController<Uint8Array>
+    const body = new ReadableStream<Uint8Array>({ start(c) { controller = c } })
+    vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(body)))
+    const stream = useStream({ url: '/test' })
+    const received: string[] = []
+    stream.on('content_delta', data => received.push(data.delta))
+    const connected = stream.connect()
+    const encoder = new TextEncoder()
+    try {
+      controller.enqueue(encoder.encode('data: {"delta":"first"}\n\n'))
+      await vi.waitFor(() => expect(received).toEqual(['first']))
+      controller.enqueue(encoder.encode('data: {"delta":"second"}\r\n\r\n'))
+      await vi.waitFor(() => expect(received).toEqual(['first', 'second']))
+    } finally {
+      controller.close()
+      await connected
+    }
+  })
+})

mateclaw-ui/src/composables/chat/useStream.ts

@@ -147,9 +147,16 @@ export interface UseStreamReturn {
 class SSEParser {
   private buffer = ''
   private readonly separator = '\n\n'
+  private trailingCR = false
 
   parse(chunk: string): SSEEvent[] {
-    this.buffer += chunk
+    // SSE permits LF, CRLF and CR. Normalize before framing, including a
+    // CRLF pair split across network reads, without manufacturing blank lines.
+    if (chunk.length > 0) {
+      const skipLF = this.trailingCR && chunk.startsWith('\n')
+      this.trailingCR = chunk.endsWith('\r')
+      this.buffer += (skipLF ? chunk.slice(1) : chunk).replace(/\r\n?/g, '\n')
+    }
     const events: SSEEvent[] = []
     
     // 分割事件块
@@ -179,8 +186,7 @@ class SSEParser {
   private parseEvent(part: string): SSEEvent | null {
     const lines = part.split('\n')
     let eventType: SSEEventType = 'content_delta'
-    let data: any = {}
-    let hasData = false
+    const dataLines: string[] = []
     let eventId: string | undefined
 
     for (const line of lines) {
@@ -189,24 +195,27 @@ class SSEParser {
       const colonIndex = line.indexOf(':')
       if (colonIndex === -1) continue
 
-      const key = line.slice(0, colonIndex).trim()
-      const value = line.slice(colonIndex + 1).trim()
+      const key = line.slice(0, colonIndex)
+      const rawValue = line.slice(colonIndex + 1)
+      const value = rawValue.startsWith(' ') ? rawValue.slice(1) : rawValue
 
       if (key === 'event') {
         eventType = value as SSEEventType
       } else if (key === 'id') {
         eventId = value
       } else if (key === 'data') {
-        hasData = true
-        try {
-          data = JSON.parse(value)
-        } catch {
-          data = value
-        }
+        dataLines.push(value)
       }
     }
 
-    if (!hasData) return null
+    if (dataLines.length === 0) return null
+    const rawData = dataLines.join('\n')
+    let data: any
+    try {
+      data = JSON.parse(rawData)
+    } catch {
+      data = rawData
+    }
     return eventId !== undefined
       ? { type: eventType, data, id: eventId }
       : { type: eventType, data }