fix(client): handle stream start rejection in subscribe

packages/typescript-client/src/client.ts

@@ -1825,7 +1825,12 @@ export class ShapeStream<T extends Row<unknown> = Row>
     const subscriptionId = {}
 
     this.#subscribers.set(subscriptionId, [callback, onError])
-    if (!this.#started) this.#start()
+    if (!this.#started) {
+      this.#start().catch(() => {
+        // Errors from #start are handled internally via onError.
+        // This catch prevents unhandled promise rejection in Node/Bun.
+      })
+    }
 
     return () => {
       this.#subscribers.delete(subscriptionId)

packages/typescript-client/test/stream.test.ts

@@ -1,6 +1,7 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   ShapeStream,
+  FetchError,
   isChangeMessage,
   isControlMessage,
   Message,
@@ -25,6 +26,36 @@ describe(`ShapeStream`, () => {
 
   afterEach(() => aborter.abort())
 
+  it(`does not create an unhandled rejection when a subscriber handles an error`, async () => {
+    const unhandledRejections: unknown[] = []
+    const onUnhandledRejection = (reason: unknown) => {
+      unhandledRejections.push(reason)
+    }
+    process.on(`unhandledRejection`, onUnhandledRejection)
+
+    try {
+      const stream = new ShapeStream({
+        url: shapeUrl,
+        params: { table: `test` },
+        signal: aborter.signal,
+        fetchClient: async () =>
+          new Response(undefined, {
+            status: 401,
+          }),
+      })
+
+      const subscriberError = new Promise<Error>((resolve) => {
+        stream.subscribe(() => {}, resolve)
+      })
+
+      await expect(subscriberError).resolves.toBeInstanceOf(FetchError)
+      await resolveInMacrotask(undefined)
+      expect(unhandledRejections).toEqual([])
+    } finally {
+      process.off(`unhandledRejection`, onUnhandledRejection)
+    }
+  })
+
   it(`requestSnapshot waits for snapshot messages to be published to subscribers before resolving`, async () => {
     const snapshotRow = {
       key: `test-1`,