fix(ai-client): port #508 robustness fixes onto fetcher-alt branch

The fetcher path uses the same SSE parsing and connect-wrapper plumbing
as the stream() path on #508, so the polish that landed during #508's
review applies directly here. Carry it over so this branch has the same
robustness.

- Skip SSE control lines (`:` comments, `event:` / `id:` / `retry:`) in
  responseToSSEChunks. Proxies and CDNs inject these as keepalives;
  letting them through would feed JSON.parse a non-payload line.
- Drop unterminated trailing buffer in readStreamLines. A non-empty
  buffer at stream end means the connection was cut mid-line, so the
  data is partial — yielding it would surface a misleading RUN_ERROR
  for what is really a transport-layer issue.
- Surface JSON.parse failures in responseToSSEChunks and fetchHttpStream.
  Stop swallowing them behind console.warn; let SyntaxError propagate so
  the connect-wrapper turns it into a visible RUN_ERROR.
- Drop unsafe `as unknown as StreamChunk` casts in
  normalizeConnectionAdapter's synthesized RUN_FINISHED / RUN_ERROR
  events. Use EventType + RunFinishedEvent / RunErrorEvent so missing
  required fields are caught by the compiler. Track upstream
  threadId/runId from chunks and reuse them in the synthesis instead of
  fabricating both ids unconditionally.
- Forward optional abortSignal third arg through stream() and rpcStream()
  factory signatures. Backwards-compatible for existing callers; lets
  long-running factories cancel when useChat aborts. Mirrors what
  fetcherToConnectionAdapter already does.

Tests:
- Update the two `should handle malformed JSON gracefully` tests to
  assert SyntaxError throws instead of silent drop.
- Update stream() / rpcStream() factory mock assertions to expect the new
  third arg.
- Add chat-fetcher test asserting a fetcher returning a malformed-SSE
  Response surfaces as a RUN_ERROR via onError.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: tombeckenham

packages/typescript/ai-client/src/connection-adapters.ts

@@ -1,4 +1,11 @@
-import type { ModelMessage, StreamChunk, UIMessage } from '@tanstack/ai'
+import { EventType } from '@tanstack/ai'
+import type {
+  ModelMessage,
+  RunErrorEvent,
+  RunFinishedEvent,
+  StreamChunk,
+  UIMessage,
+} from '@tanstack/ai'
 import type { ChatFetcher } from './types'
 
 /**
@@ -54,9 +61,14 @@ async function* readStreamLines(
       }
     }
 
-    // Process any remaining data in the buffer
+    // Drop any unterminated trailing buffer. A non-empty buffer at stream end
+    // means the connection was cut mid-line (server crash, dropped TCP), so
+    // the content is by definition partial — yielding it would feed truncated
+    // JSON to downstream parsers and produce a confusing RUN_ERROR.
     if (buffer.trim()) {
-      yield buffer
+      console.warn(
+        '[@tanstack/ai-client] Stream ended with unterminated trailing data; discarding. The connection was likely cut short.',
+      )
     }
   } finally {
     reader.releaseLock()
@@ -68,6 +80,14 @@ async function* readStreamLines(
  *
  * Used by both `fetchServerSentEvents` (HTTP path) and
  * `fetcherToConnectionAdapter` (when a fetcher returns a Response).
+ *
+ * Accepts either `data: {...}` SSE lines or bare JSON lines (legacy/raw mode).
+ * Skips non-payload SSE fields (comments starting with `:`, and `event:` /
+ * `id:` / `retry:` lines) — proxies and CDNs may inject these as keepalives,
+ * and they are not malformed JSON.
+ *
+ * A JSON parse failure on an actual payload line throws (surfacing as
+ * RUN_ERROR through the connect-wrapper) rather than being silently dropped.
  */
 async function* responseToSSEChunks(
   response: Response,
@@ -83,20 +103,22 @@ async function* responseToSSEChunks(
     throw new Error('Response body is not readable')
   }
   for await (const line of readStreamLines(reader, abortSignal)) {
+    if (
+      line.startsWith(':') ||
+      line.startsWith('event:') ||
+      line.startsWith('id:') ||
+      line.startsWith('retry:')
+    ) {
+      continue
+    }
     const data = line.startsWith('data: ') ? line.slice(6) : line
-
     if (data === '[DONE]') {
       console.warn(
         '[@tanstack/ai-client] Received [DONE] sentinel. This is deprecated — upgrade your @tanstack/ai server package. RUN_FINISHED is the stream terminator.',
       )
       continue
     }
-
-    try {
-      yield JSON.parse(data) as StreamChunk
-    } catch (parseError) {
-      console.warn('Failed to parse SSE chunk:', data)
-    }
+    yield JSON.parse(data) as StreamChunk
   }
 }
 
@@ -213,9 +235,17 @@ export function normalizeConnectionAdapter(
     },
     async send(messages, data, abortSignal) {
       let hasTerminalEvent = false
+      let upstreamThreadId: string | undefined
+      let upstreamRunId: string | undefined
       try {
         const stream = connection.connect(messages, data, abortSignal)
         for await (const chunk of stream) {
+          if ('threadId' in chunk && typeof chunk.threadId === 'string') {
+            upstreamThreadId = chunk.threadId
+          }
+          if ('runId' in chunk && typeof chunk.runId === 'string') {
+            upstreamRunId = chunk.runId
+          }
           if (chunk.type === 'RUN_FINISHED' || chunk.type === 'RUN_ERROR') {
             hasTerminalEvent = true
           }
@@ -225,28 +255,26 @@ export function normalizeConnectionAdapter(
         // If the connect stream ended cleanly without a terminal event,
         // synthesize RUN_FINISHED so request-scoped consumers can complete.
         if (!abortSignal?.aborted && !hasTerminalEvent) {
-          push({
-            type: 'RUN_FINISHED',
-            runId: `run-${Date.now()}`,
+          const synthetic: RunFinishedEvent = {
+            type: EventType.RUN_FINISHED,
+            threadId: upstreamThreadId ?? `thread-${Date.now()}`,
+            runId: upstreamRunId ?? `run-${Date.now()}`,
             model: 'connect-wrapper',
             timestamp: Date.now(),
             finishReason: 'stop',
-          } as unknown as StreamChunk)
+          }
+          push(synthetic)
         }
       } catch (err) {
         if (!abortSignal?.aborted && !hasTerminalEvent) {
-          push({
-            type: 'RUN_ERROR',
+          const message =
+            err instanceof Error ? err.message : 'Unknown error in connect()'
+          const synthetic: RunErrorEvent = {
+            type: EventType.RUN_ERROR,
             timestamp: Date.now(),
-            message:
-              err instanceof Error ? err.message : 'Unknown error in connect()',
-            error: {
-              message:
-                err instanceof Error
-                  ? err.message
-                  : 'Unknown error in connect()',
-            },
-          } as unknown as StreamChunk)
+            message,
+          }
+          push(synthetic)
         }
         throw err
       }
@@ -421,12 +449,7 @@ export function fetchHttpStream(
       }
 
       for await (const line of readStreamLines(reader, abortSignal)) {
-        try {
-          const parsed: StreamChunk = JSON.parse(line)
-          yield parsed
-        } catch (parseError) {
-          console.warn('Failed to parse HTTP stream chunk:', line)
-        }
+        yield JSON.parse(line) as StreamChunk
       }
     },
   }
@@ -450,13 +473,14 @@ export function stream(
   streamFactory: (
     messages: Array<UIMessage> | Array<ModelMessage>,
     data?: Record<string, any>,
+    abortSignal?: AbortSignal,
   ) => AsyncIterable<StreamChunk>,
 ): ConnectConnectionAdapter {
   return {
-    async *connect(messages, data) {
+    async *connect(messages, data, abortSignal) {
       // Pass messages as-is (UIMessages with parts preserved)
       // Server-side chat() handles conversion to ModelMessages
-      yield* streamFactory(messages, data)
+      yield* streamFactory(messages, data, abortSignal)
     },
   }
 }
@@ -520,13 +544,14 @@ export function rpcStream(
   rpcCall: (
     messages: Array<UIMessage> | Array<ModelMessage>,
     data?: Record<string, any>,
+    abortSignal?: AbortSignal,
   ) => AsyncIterable<StreamChunk>,
 ): ConnectConnectionAdapter {
   return {
-    async *connect(messages, data) {
+    async *connect(messages, data, abortSignal) {
       // Pass messages as-is (UIMessages with parts preserved)
       // Server-side chat() handles conversion to ModelMessages
-      yield* rpcCall(messages, data)
+      yield* rpcCall(messages, data, abortSignal)
     },
   }
 }

packages/typescript/ai-client/tests/chat-fetcher.test.ts

@@ -131,6 +131,34 @@ describe('ChatClient — fetcher transport', () => {
     expect(client.getStatus()).toBe('error')
   })
 
+  it('surfaces a malformed-SSE Response as a ChatClient error', async () => {
+    // A fetcher that returns a Response whose body has a malformed JSON line.
+    // The new behavior is to throw SyntaxError from the SSE parser; the
+    // chat client should surface that as an error rather than silently
+    // dropping the bad chunk.
+    const sseBody = 'data: { not valid json\n\n'
+    const fetcher: ChatFetcher = async () => {
+      return new Response(sseBody, {
+        status: 200,
+        headers: { 'content-type': 'text/event-stream' },
+      })
+    }
+
+    let observedError: Error | undefined
+    const client = new ChatClient({
+      fetcher,
+      onError: (err) => {
+        observedError = err
+      },
+    })
+
+    await client.sendMessage('hi')
+
+    expect(observedError).toBeDefined()
+    expect(observedError!.name).toBe('SyntaxError')
+    expect(client.getStatus()).toBe('error')
+  })
+
   it('passes UIMessages and merged body to the fetcher', async () => {
     const fetcher = vi.fn<ChatFetcher>(async function* () {
       yield {

packages/typescript/ai-client/tests/connection-adapters.test.ts

@@ -144,11 +144,7 @@ describe('connection-adapters', () => {
       warnSpy.mockRestore()
     })
 
-    it('should handle malformed JSON gracefully', async () => {
-      const consoleWarnSpy = vi
-        .spyOn(console, 'warn')
-        .mockImplementation(() => {})
-
+    it('should throw a SyntaxError on malformed JSON', async () => {
       const mockReader = {
         read: vi
           .fn()
@@ -170,17 +166,16 @@ describe('connection-adapters', () => {
       fetchMock.mockResolvedValue(mockResponse as any)
 
       const adapter = fetchServerSentEvents('/api/chat')
-      const chunks: Array<StreamChunk> = []
 
-      for await (const chunk of adapter.connect([
-        { role: 'user', content: 'Hello' },
-      ])) {
-        chunks.push(chunk)
-      }
-
-      expect(chunks).toHaveLength(0)
-      expect(consoleWarnSpy).toHaveBeenCalled()
-      consoleWarnSpy.mockRestore()
+      await expect(
+        (async () => {
+          for await (const _ of adapter.connect([
+            { role: 'user', content: 'Hello' },
+          ])) {
+            // Consume
+          }
+        })(),
+      ).rejects.toThrow(SyntaxError)
     })
 
     it('should handle HTTP errors', async () => {
@@ -550,11 +545,7 @@ describe('connection-adapters', () => {
       expect(chunks).toHaveLength(1)
     })
 
-    it('should handle malformed JSON gracefully', async () => {
-      const consoleWarnSpy = vi
-        .spyOn(console, 'warn')
-        .mockImplementation(() => {})
-
+    it('should throw a SyntaxError on malformed JSON', async () => {
       const mockReader = {
         read: vi
           .fn()
@@ -576,17 +567,16 @@ describe('connection-adapters', () => {
       fetchMock.mockResolvedValue(mockResponse as any)
 
       const adapter = fetchHttpStream('/api/chat')
-      const chunks: Array<StreamChunk> = []
 
-      for await (const chunk of adapter.connect([
-        { role: 'user', content: 'Hello' },
-      ])) {
-        chunks.push(chunk)
-      }
-
-      expect(chunks).toHaveLength(0)
-      expect(consoleWarnSpy).toHaveBeenCalled()
-      consoleWarnSpy.mockRestore()
+      await expect(
+        (async () => {
+          for await (const _ of adapter.connect([
+            { role: 'user', content: 'Hello' },
+          ])) {
+            // Consume
+          }
+        })(),
+      ).rejects.toThrow(SyntaxError)
     })
 
     it('should handle HTTP errors', async () => {
@@ -836,6 +826,7 @@ describe('connection-adapters', () => {
       expect(streamFactory).toHaveBeenCalledWith(
         expect.arrayContaining([expect.objectContaining({ role: 'user' })]),
         data,
+        undefined,
       )
     })
   })
@@ -1023,6 +1014,7 @@ describe('connection-adapters', () => {
       expect(rpcCall).toHaveBeenCalledWith(
         expect.arrayContaining([expect.objectContaining({ role: 'user' })]),
         data,
+        undefined,
       )
     })
   })