fix(responses): reject with OpenAIError when the stream emits an error event

src/lib/responses/ResponseStream.ts

@@ -8,7 +8,7 @@ import type {
 } from '../../resources/responses/responses';
 import type { RequestOptions } from '../../internal/request-options';
 import type { ReadableStream } from '../../internal/shim-types';
-import { APIUserAbortError, OpenAIError } from '../../error';
+import { APIError, APIUserAbortError, OpenAIError } from '../../error';
 import type OpenAI from '../../index';
 import { EventStream } from '../EventStream';
 import type { BaseEvents } from '../EventStream';
@@ -69,7 +69,7 @@ type ResponseEvents = BaseEvents &
         >,
       ) => void;
     },
-    'response.output_text.delta' | 'response.function_call_arguments.delta'
+    'response.output_text.delta' | 'response.function_call_arguments.delta' | 'error'
   > & {
     /** Called for every raw response event that passes the replay sequence filter. */
     event: (event: ResponseStreamEvent) => void;
@@ -141,6 +141,14 @@ export class ResponseStream<ParsedT = null>
       }
     };
 
+    if (event.type === 'error') {
+      // First-party providers nest their error payload; retain flat compatibility for
+      // serialized events matching the currently published event schema.
+      const error =
+        'error' in event && typeof event.error === 'object' && event.error !== null ? event.error : event;
+      throw new APIError(undefined, error, event.message, undefined);
+    }
+
     const response = accumulateResponse(event, this.#currentResponseSnapshot);
     this.#currentResponseSnapshot = response;
     maybeEmit('event', event);
@@ -257,58 +265,11 @@ export class ResponseStream<ParsedT = null>
 
   /** Iterates over response events; stopping iteration early aborts the underlying request. */
   [Symbol.asyncIterator](this: ResponseStream<ParsedT>): AsyncIterator<ResponseStreamEvent> {
-    const pushQueue: ResponseStreamEvent[] = [];
-    const readQueue: {
-      resolve: (event: ResponseStreamEvent | undefined) => void;
-      reject: (err: unknown) => void;
-    }[] = [];
-    let done = false;
-
-    this.on('event', (event) => {
-      const reader = readQueue.shift();
-      if (reader) {
-        reader.resolve(event);
-      } else {
-        pushQueue.push(event);
-      }
-    });
-
-    this.on('end', () => {
-      done = true;
-      for (const reader of readQueue) {
-        reader.resolve(undefined);
-      }
-      readQueue.length = 0;
-    });
-
-    this.on('abort', (err) => {
-      done = true;
-      for (const reader of readQueue) {
-        reader.reject(err);
-      }
-      readQueue.length = 0;
-    });
-
-    this.on('error', (err) => {
-      done = true;
-      for (const reader of readQueue) {
-        reader.reject(err);
-      }
-      readQueue.length = 0;
-    });
-
+    const iterator = this.events('event');
     return {
-      next: async (): Promise<IteratorResult<ResponseStreamEvent>> => {
-        if (!pushQueue.length) {
-          if (done) {
-            return { value: undefined, done: true };
-          }
-          return new Promise<ResponseStreamEvent | undefined>((resolve, reject) =>
-            readQueue.push({ resolve, reject }),
-          ).then((event) => (event ? { value: event, done: false } : { value: undefined, done: true }));
-        }
-        const event = pushQueue.shift()!;
-        return { value: event, done: false };
+      next: async () => {
+        const result = await iterator.next();
+        return result.done ? { value: undefined, done: true } : { value: result.value[0], done: false };
       },
       return: async () => {
         this.abort();