feat(client, server): add eventIteratorToUnproxiedDataStream util (#1110)

Some event iterators proxy their emitted data to track event metadata.
When integrating with tools that don't support proxied data (such as AI
SDK), use eventIteratorToUnproxiedDataStream instead of
eventIteratorToStream to ensure raw, unproxied data is emitted.

Fixed: #977

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Added a data-stream conversion that returns unproxied values for
streamed results, exposed across client and server packages for better
compatibility with AI SDKs and integrations.

* **Documentation**
* Updated AI SDK integration docs to recommend using the new unproxied
data-stream handling for reliable structured-data processing.

* **Tests**
* Added tests covering normal, empty, error, and cancellation scenarios
for the new data-stream behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: dinwwwh

apps/content/docs/integrations/ai-sdk.md

@@ -59,13 +59,13 @@ export const chat = os
 const client = { chat }
 // ---cut---
 import { useChat } from '@ai-sdk/react'
-import { eventIteratorToStream } from '@orpc/client'
+import { eventIteratorToUnproxiedDataStream } from '@orpc/client'
 
 export function Example() {
   const { messages, sendMessage, status } = useChat({
     transport: {
       async sendMessages(options) {
-        return eventIteratorToStream(await client.chat({
+        return eventIteratorToUnproxiedDataStream(await client.chat({
           chatId: options.chatId,
           messages: options.messages,
         }, { signal: options.abortSignal }))
@@ -115,3 +115,9 @@ export function Example() {
 ::: info
 The `reconnectToStream` function is not supported by default, which is fine for most use cases. If you need reconnection support, implement it similar to `sendMessages` with custom reconnection logic. See this [reconnect example](<https://github.com/vercel/ai-chatbot/blob/main/app/(chat)/api/chat/%5Bid%5D/stream/route.ts>).
 :::
+
+::: info
+Prefer `eventIteratorToUnproxiedDataStream` over `eventIteratorToStream`.
+AI SDK internally uses `structuredClone`, which doesn't support proxied data.
+oRPC may proxy events for metadata, so unproxy before passing to AI SDK.
+:::

packages/client/src/index.ts

@@ -10,6 +10,7 @@ export * from './utils'
 export {
   AsyncIteratorClass,
   asyncIteratorToStream as eventIteratorToStream,
+  asyncIteratorToUnproxiedDataStream as eventIteratorToUnproxiedDataStream,
   EventPublisher,
   onError,
   onFinish,

packages/server/src/index.ts

@@ -43,6 +43,7 @@ export type { IntersectPick } from '@orpc/shared'
 export {
   AsyncIteratorClass,
   asyncIteratorToStream as eventIteratorToStream,
+  asyncIteratorToUnproxiedDataStream as eventIteratorToUnproxiedDataStream,
   EventPublisher,
   onError,
   onFinish,

packages/shared/src/stream.test.ts

@@ -1,5 +1,5 @@
 import { AsyncIteratorClass } from './iterator'
-import { asyncIteratorToStream, streamToAsyncIteratorClass } from './stream'
+import { asyncIteratorToStream, asyncIteratorToUnproxiedDataStream, streamToAsyncIteratorClass } from './stream'
 
 describe('streamToAsyncIteratorClass', () => {
   it('should convert a ReadableStream to AsyncIteratorClass', async () => {
@@ -187,3 +187,121 @@ it('streamToAsyncIteratorClass + asyncIteratorToStream', async () => {
 
   expect(results).toEqual([1, 2, 3])
 })
+
+describe('asyncIteratorToUnproxiedDataStream', () => {
+  const PROXY_SYMBOL = Symbol('proxy')
+
+  function proxy(value: object) {
+    return new Proxy(value, {
+      get(target, prop) {
+        if (prop === PROXY_SYMBOL) {
+          return true
+        }
+
+        return Reflect.get(target, prop)
+      },
+    })
+  }
+
+  function isProxied(value: any) {
+    return Boolean(typeof value === 'object' && value && value[PROXY_SYMBOL])
+  }
+
+  it('should convert an AsyncIterator to ReadableStream and unproxied data', async () => {
+    const date = new Date()
+    const set = new Set([date])
+
+    async function* generator() {
+      yield 1
+      yield proxy({ order: 2 })
+      yield { order: 3 }
+      yield [4]
+      yield proxy([5])
+      yield date // support native Date
+      yield set // support native Set
+    }
+
+    const asyncIterator = generator()
+    const stream = asyncIteratorToUnproxiedDataStream(asyncIterator)
+
+    expect(stream).toBeInstanceOf(ReadableStream)
+
+    const reader = stream.getReader()
+    const results: any[] = []
+
+    let result = await reader.read()
+    while (!result.done) {
+      results.push(result.value)
+      result = await reader.read()
+    }
+
+    expect(results).toEqual([
+      1,
+      { order: 2 },
+      { order: 3 },
+      [4],
+      [5],
+      date,
+      set,
+    ])
+
+    expect(results.some(isProxied)).toBe(false)
+  })
+
+  it('should handle empty async iterator', async () => {
+    async function* emptyGenerator() {
+      // Empty generator
+    }
+
+    const asyncIterator = emptyGenerator()
+    const stream = asyncIteratorToUnproxiedDataStream(asyncIterator)
+
+    const reader = stream.getReader()
+    const result = await reader.read()
+
+    expect(result.done).toBe(true)
+    expect(result.value).toBeUndefined()
+  })
+
+  it('should handle async iterator errors', async () => {
+    const error = new Error('Iterator error')
+    async function* errorGenerator() {
+      yield 1
+      throw error
+    }
+
+    const asyncIterator = errorGenerator()
+    const stream = asyncIteratorToUnproxiedDataStream(asyncIterator)
+
+    const reader = stream.getReader()
+
+    // First read should succeed
+    const firstResult = await reader.read()
+    expect(firstResult.done).toBe(false)
+    expect(firstResult.value).toBe(1)
+
+    // Second read should throw the error
+    await expect(reader.read()).rejects.toThrow(error)
+  })
+
+  it('should call iterator.return when stream is cancelled', async () => {
+    let cleanupCalled = false
+
+    const stream = asyncIteratorToUnproxiedDataStream(async function* () {
+      try {
+        yield 1
+        yield 2
+        await new Promise(resolve => setTimeout(resolve, 100)) // Simulate async operation
+      }
+      finally {
+        cleanupCalled = true
+      }
+    }())
+
+    const reader = stream.getReader()
+    await reader.read()
+    await reader.cancel()
+
+    expect(cleanupCalled).toBe(true)
+  })
+})

packages/shared/src/stream.ts

@@ -1,5 +1,9 @@
 import { AsyncIteratorClass } from './iterator'
+import { isObject } from './object'
 
+/**
+ * Converts a `ReadableStream` into an `AsyncIteratorClass`.
+ */
 export function streamToAsyncIteratorClass<T>(
   stream: ReadableStream<T>,
 ): AsyncIteratorClass<T> {
@@ -15,6 +19,9 @@ export function streamToAsyncIteratorClass<T>(
   )
 }
 
+/**
+ * Converts an `AsyncIterator` into a `ReadableStream`.
+ */
 export function asyncIteratorToStream<T>(
   iterator: AsyncIterator<T>,
 ): ReadableStream<T> {
@@ -34,3 +41,33 @@ export function asyncIteratorToStream<T>(
     },
   })
 }
+
+/**
+ * Converts an `AsyncIterator` into a `ReadableStream`, ensuring that
+ * all emitted object values are *unproxied* before enqueuing.
+ */
+export function asyncIteratorToUnproxiedDataStream<T>(
+  iterator: AsyncIterator<T>,
+): ReadableStream<T> {
+  return new ReadableStream<T>({
+    async pull(controller) {
+      const { done, value } = await iterator.next()
+
+      if (done) {
+        controller.close()
+      }
+      else {
+        const unproxied = isObject(value)
+          ? { ...value }
+          : Array.isArray(value)
+            ? value.map(i => i) as T // use .map instead of ... to deal with sparse arrays
+            : value
+
+        controller.enqueue(unproxied)
+      }
+    },
+    async cancel() {
+      await iterator.return?.()
+    },
+  })
+}