fix(stream): usage on finalChatCompletion, fence-strip in .text, APIUserAbortError on abort

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<ChatCompleti
   #id = "";
   #model = "";
   #created = 0;
+  #usage: ChatCompletionChunk["usage"];
+  #systemFingerprint: string | undefined;
   #toolCalls = new Map<number, ToolCallAcc>();
 
   constructor(openai: OpenAI, body: Record<string, unknown>, options?: RequestOptions, stripFence = false) {
@@ -52,14 +55,24 @@ export class InterfazeChatCompletionStream implements AsyncIterable<ChatCompleti
     return this.#raw;
   }
 
+  #throwIfAborted(): void {
+    if (this.#options?.signal?.aborted) throw new APIUserAbortError();
+  }
+
   async *[Symbol.asyncIterator](): AsyncIterator<ChatCompletionChunk> {
     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<ChatCompleti
     this.#started = true;
     const filter = new SideChannelFilter();
     const raw = await this.#getRaw();
-    for await (const chunk of raw) {
-      this.#accumulate(chunk);
-      const piece = chunk.choices?.[0]?.delta?.content;
-      if (typeof piece === "string" && piece) {
-        const visible = filter.feed(piece);
-        if (visible) yield visible;
+    try {
+      for await (const chunk of raw) {
+        this.#accumulate(chunk);
+        const piece = chunk.choices?.[0]?.delta?.content;
+        if (typeof piece === "string" && piece) {
+          const visible = filter.feed(piece);
+          if (visible) yield visible;
+        }
       }
+    } catch (err) {
+      this.#throwIfAborted();
+      throw err;
     }
+    this.#throwIfAborted();
     const tail = filter.flush();
     if (tail) yield tail;
     this.#done = true;
@@ -90,6 +109,8 @@ export class InterfazeChatCompletionStream implements AsyncIterable<ChatCompleti
     if (!this.#id && chunk.id) this.#id = chunk.id;
     if (!this.#model && chunk.model) this.#model = chunk.model;
     if (!this.#created && chunk.created) this.#created = chunk.created;
+    if (chunk.usage) this.#usage = chunk.usage;
+    if (chunk.system_fingerprint) this.#systemFingerprint = chunk.system_fingerprint;
     const choice = chunk.choices?.[0];
     if (!choice) return;
     const delta = choice.delta;
@@ -105,17 +126,24 @@ export class InterfazeChatCompletionStream implements AsyncIterable<ChatCompleti
     }
   }
 
-  /** Concatenated visible content (side-channel blocks removed). */
+  /** Concatenated visible content (side-channel blocks removed; json_object fence unwrapped). */
   get text(): string {
-    return stripSideChannels(this.#content).text;
+    const t = stripSideChannels(this.#content).text;
+    return this.#stripFence ? stripJsonFence(t) : t;
   }
 
   /** Drive the stream to completion (if not already) and return the assembled completion. */
   async finalChatCompletion(): Promise<InterfazeChatCompletion> {
     if (!this.#started) {
       this.#started = true;
       const raw = await this.#getRaw();
-      for await (const chunk of raw) this.#accumulate(chunk);
+      try {
+        for await (const chunk of raw) this.#accumulate(chunk);
+      } catch (err) {
+        this.#throwIfAborted();
+        throw err;
+      }
+      this.#throwIfAborted();
       this.#done = true;
     } else if (!this.#done) {
       throw new InterfazeError("Call finalChatCompletion() after fully iterating the stream, or instead of iterating.");
@@ -147,6 +175,8 @@ export class InterfazeChatCompletionStream implements AsyncIterable<ChatCompleti
     } as unknown as InterfazeChatCompletion;
     if (reasoning) completion.reasoning = reasoning;
     if (precontext) completion.precontext = precontext;
+    if (this.#usage) completion.usage = this.#usage;
+    if (this.#systemFingerprint) completion.system_fingerprint = this.#systemFingerprint;
     return completion;
   }
 }

test/stream.test.ts

@@ -1,3 +1,4 @@
+import { APIUserAbortError } from "openai";
 import { describe, expect, it } from "vitest";
 import { fixture, mockInterfaze, sseResponse } from "./helpers.js";
 
@@ -26,6 +27,20 @@ const fencedJson = [
   mkChunk({}, "stop"),
 ];
 
+const usageChunks = [
+  mkChunk({ content: "Hi" }),
+  mkChunk({}, "stop"),
+  {
+    id: "req-x",
+    object: "chat.completion.chunk",
+    created: 1,
+    model: "interfaze-beta",
+    choices: [],
+    usage: { prompt_tokens: 11, completion_tokens: 2, total_tokens: 13 },
+    system_fingerprint: "fp_test",
+  },
+];
+
 describe("streaming accumulator", () => {
   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);
+  });
 });