From 6e1b02dce51f98fa70d853aa17b55c7801a20b1d Mon Sep 17 00:00:00 2001
From: huyengiang101086-spec
 <270684046+huyengiang101086-spec@users.noreply.github.com>
Date: Mon, 12 Oct 2026 21:41:07 +0800
Subject: [PATCH] fix(cli): observe SSE read failures while consumers process
 events

---
 cli/src/call/sse.test.ts | 42 ++++++++++++++++++++++++++++++++++++++++
 cli/src/call/sse.ts      |  4 ++++
 2 files changed, 46 insertions(+)

diff --git a/cli/src/call/sse.test.ts b/cli/src/call/sse.test.ts
index 4a901c09dd80a0..d80b2c0e6e3df4 100644
--- a/cli/src/call/sse.test.ts
+++ b/cli/src/call/sse.test.ts
@@ -1,3 +1,5 @@
+import { execFile } from 'node:child_process'
+import { promisify } from 'node:util'
 import { describe, expect, it } from 'vite-plus/test'
 import { parseSSE } from './sse'
 
@@ -66,6 +68,46 @@ describe('parseSSE', () => {
     expect((await collect(s))[0]?.data).toBe('😀')
   })
 
+  it('propagates a response body error to the consumer', async () => {
+    const error = new Error('network disconnected')
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.error(error)
+      },
+    })
+    await expect(collect(body)).rejects.toBe(error)
+  })
+
+  it('observes read failures while the consumer is processing a yielded event', async () => {
+    // Isolate Node's strict unhandled-rejection policy from the test runner.
+    const source = new URL('./sse.ts', import.meta.url).href
+    const script = `
+      import assert from 'node:assert/strict'
+      import { setImmediate } from 'node:timers/promises'
+      import { parseSSE } from ${JSON.stringify(source)}
+      let controller
+      const body = new ReadableStream({ start(c) { controller = c } })
+      const events = parseSSE(body)
+      controller.enqueue(new TextEncoder().encode('data: first\\n\\ndata: second\\n\\n'))
+      const decoder = new TextDecoder()
+      assert.equal(decoder.decode((await events.next()).value.data), 'first')
+      const error = new Error('network disconnected')
+      controller.error(error)
+      await setImmediate()
+      assert.equal(decoder.decode((await events.next()).value.data), 'second')
+      await assert.rejects(events.next(), (failure) => failure === error)
+      console.log('stream error reached the consumer')
+    `
+    const { stdout, stderr } = await promisify(execFile)(process.execPath, [
+      '--unhandled-rejections=strict',
+      '--input-type=module',
+      '-e',
+      script,
+    ])
+    expect(stdout.trim()).toBe('stream error reached the consumer')
+    expect(stderr).toBe('')
+  })
+
   it('aborts when signal fires', async () => {
     const ctrl = new AbortController()
     const slow = new ReadableStream<Uint8Array>({
diff --git a/cli/src/call/sse.ts b/cli/src/call/sse.ts
index 29d441a177a516..c60f4c6dd3a0fc 100644
--- a/cli/src/call/sse.ts
+++ b/cli/src/call/sse.ts
@@ -68,6 +68,10 @@ export async function* parseSSE(
     }
   })()
 
+  // Observe background failures even while a yielded event suspends the consumer.
+  // Awaiting the original pump below still propagates the error after the queue drains.
+  void pump.catch(() => {})
+
   try {
     while (true) {
       while (queue.length > 0) {