diff --git a/src/stream.ts b/src/stream.ts index b97e157..794ae6a 100644 --- a/src/stream.ts +++ b/src/stream.ts @@ -1,4 +1,5 @@ import type OpenAI from "openai"; +import { APIUserAbortError } from "openai"; import type { ChatCompletionChunk } from "openai/resources/chat/completions/completions"; import { InterfazeError } from "./errors.js"; @@ -33,6 +34,8 @@ export class InterfazeChatCompletionStream implements AsyncIterable(); constructor(openai: OpenAI, body: Record, options?: RequestOptions, stripFence = false) { @@ -52,14 +55,24 @@ export class InterfazeChatCompletionStream implements AsyncIterable { if (this.#started) throw new InterfazeError("This stream has already been consumed."); this.#started = true; const raw = await this.#getRaw(); - for await (const chunk of raw) { - this.#accumulate(chunk); - yield chunk; + try { + for await (const chunk of raw) { + this.#accumulate(chunk); + yield chunk; + } + } catch (err) { + this.#throwIfAborted(); + throw err; } + this.#throwIfAborted(); this.#done = true; } @@ -73,14 +86,20 @@ export class InterfazeChatCompletionStream implements AsyncIterable { it("iterates role-less chunks without throwing 'missing role'", async () => { const { interfaze } = mockInterfaze(() => sseResponse(streamBasic)); @@ -139,4 +154,42 @@ describe("streaming accumulator", () => { const content = (await s.finalChatCompletion()).choices[0]!.message.content!; expect(content.startsWith("```")).toBe(true); }); + + it("finalChatCompletion surfaces streamed usage and system_fingerprint", async () => { + const { interfaze } = mockInterfaze(() => sseResponse(usageChunks)); + const s = interfaze.chat.completions.stream({ + messages: [{ role: "user", content: "x" }], + stream_options: { include_usage: true }, + }); + const final = await s.finalChatCompletion(); + expect(final.usage?.total_tokens).toBe(13); + expect(final.usage?.prompt_tokens).toBe(11); + expect(final.system_fingerprint).toBe("fp_test"); + }); + + it(".text strips the json_object fence, matching finalChatCompletion()", async () => { + const { interfaze } = mockInterfaze(() => sseResponse(fencedJson)); + const s = interfaze.chat.completions.stream({ + messages: [{ role: "user", content: "json" }], + response_format: { type: "json_object" }, + }); + const content = (await s.finalChatCompletion()).choices[0]!.message.content!; + expect(s.text.startsWith("```")).toBe(false); + expect(s.text).toBe(content); + expect(JSON.parse(s.text)).toHaveProperty("city"); + }); + + it("surfaces an aborted signal as APIUserAbortError instead of a silent end", async () => { + const controller = new AbortController(); + const { interfaze } = mockInterfaze(() => sseResponse(plainChunks)); + const s = interfaze.chat.completions.stream( + { messages: [{ role: "user", content: "x" }] }, + { signal: controller.signal }, + ); + await expect( + (async () => { + for await (const _chunk of s) controller.abort(); + })(), + ).rejects.toBeInstanceOf(APIUserAbortError); + }); });