fix: emit stream finalization errors

src/lib/EventStream.ts

@@ -44,10 +44,17 @@ export class EventStream<EventTypes extends BaseEvents> {
     // Unfortunately if we call `executor()` immediately we get runtime errors about
     // references to `this` before the `super()` constructor call returns.
     setTimeout(() => {
-      executor().then(() => {
-        this._emitFinal();
-        this._emit('end');
-      }, this.#handleError.bind(this));
+      Promise.resolve()
+        .then(executor)
+        .then(() => {
+          try {
+            this._emitFinal();
+          } catch (error) {
+            this.#handleError(error);
+            return;
+          }
+          this._emit('end');
+        }, this.#handleError.bind(this));
     }, 0);
   }
 

tests/lib/ChatCompletionStream.test.ts

@@ -8,6 +8,51 @@ import { makeStreamSnapshotRequest } from '../utils/mock-snapshots';
 jest.setTimeout(1000 * 30);
 
 describe('.stream()', () => {
+  it('emits finalization failures as errors', async () => {
+    const chunk: OpenAI.Chat.ChatCompletionChunk = {
+      id: 'chatcmpl-test',
+      object: 'chat.completion.chunk',
+      created: 1,
+      model: 'gpt-test',
+      choices: [
+        {
+          index: 0,
+          delta: { role: 'user', content: 'hello' },
+          finish_reason: 'stop',
+          logprobs: null,
+        },
+      ],
+    };
+
+    const client = {
+      chat: {
+        completions: {
+          create: jest.fn(async () => ({
+            controller: new AbortController(),
+            async *[Symbol.asyncIterator]() {
+              yield chunk;
+            },
+          })),
+        },
+      },
+    } as unknown as OpenAI;
+
+    const stream = ChatCompletionStream.createChatCompletion(client, {
+      model: 'gpt-test',
+      messages: [{ role: 'user', content: 'Say hello' }],
+    });
+    const errors: Error[] = [];
+    stream.on('error', (error) => errors.push(error));
+
+    await expect(stream.done()).rejects.toThrow(
+      'stream ended without producing a ChatCompletionMessage with role=assistant',
+    );
+    expect(errors).toHaveLength(1);
+    expect(errors[0]?.message).toBe(
+      'stream ended without producing a ChatCompletionMessage with role=assistant',
+    );
+  });
+
   it('removes the caller abort listener after the stream finishes', async () => {
     const callerController = new AbortController();
     const addEventListenerSpy = jest.spyOn(callerController.signal, 'addEventListener');