fix(cz-git): add tests and support for non-stream OpenAI chat completion responses

Author: Zhengqbbb

packages/cz-git/__tests__/stream.test.ts

@@ -0,0 +1,40 @@
+import { Readable } from 'node:stream'
+import { describe, expect, it } from 'vitest'
+import { readChatCompletionStreamToSubjects } from '../src/shared/utils/stream'
+
+function asStream(body: string): NodeJS.ReadableStream {
+    return Readable.from([body])
+}
+
+describe('readChatCompletionStreamToSubjects', () => {
+    it('parses non-stream JSON when no SSE choice deltas appear', async () => {
+        const json = JSON.stringify({
+            choices: [{ index: 0, message: { role: 'assistant', content: 'fix login redirect' } }],
+        })
+        const subjects = await readChatCompletionStreamToSubjects(asStream(json), 1)
+        expect(subjects).toEqual(['fix login redirect'])
+    })
+
+    it('parses SSE data lines as before', async () => {
+        const sse = [
+            'data: {"choices":[{"index":0,"delta":{"content":"hello"}}]}',
+            '',
+            'data: {"choices":[{"index":0,"delta":{"content":" world"}}]}',
+            '',
+            'data: [DONE]',
+            '',
+        ].join('\n')
+        const subjects = await readChatCompletionStreamToSubjects(asStream(sse), 1)
+        expect(subjects).toEqual(['hello world'])
+    })
+
+    it('throws when body has neither SSE choices nor non-stream completion', async () => {
+        await expect(readChatCompletionStreamToSubjects(asStream('not json'), 1)).rejects.toThrow(
+            /no streamed choice deltas/,
+        )
+    })
+
+    it('does not return a single empty subject when stream is empty', async () => {
+        await expect(readChatCompletionStreamToSubjects(asStream(''), 1)).rejects.toThrow()
+    })
+})

packages/cz-git/src/shared/utils/stream.ts

@@ -4,12 +4,12 @@
  * @license MIT
  */
 
-import readline from 'node:readline'
+import { Buffer } from 'node:buffer'
 import { Readable } from 'node:stream'
 import type { ReadableStream as WebReadableStream } from 'node:stream/web'
 
 /**
- * Normalize `fetch` response body to a Node.js readable stream for `readline`.
+ * Normalize `fetch` response body to a Node.js readable stream.
  */
 export function bodyToNodeReadable(body: unknown): NodeJS.ReadableStream {
     if (body == null)
@@ -47,24 +47,68 @@ export function appendVisibleDelta(acc: string, delta: { content?: unknown } | u
 
 interface StreamChoiceChunk { index?: number, delta?: { content?: unknown } }
 
+interface NonStreamChoice { index?: number, message?: { content?: unknown } }
+
+async function readableToUtf8String(stream: NodeJS.ReadableStream): Promise<string> {
+    const chunks: Buffer[] = []
+    for await (const chunk of stream as AsyncIterable<string | Buffer>) {
+        if (Buffer.isBuffer(chunk))
+            chunks.push(chunk)
+        else if (typeof chunk === 'string')
+            chunks.push(Buffer.from(chunk))
+        else
+            chunks.push(Buffer.from(String(chunk)))
+    }
+    return Buffer.concat(chunks as readonly Uint8Array[]).toString('utf8')
+}
+
 /**
- * Read an SSE stream and return one finished string per completion choice.
- * Buckets by `choices[].index` up to `choiceCount` (requested `n`).
- * Returned length matches how many indices actually appeared in the stream (capped by `choiceCount`),
- * mirroring non-stream `json.choices.length` when the provider returns fewer parallel completions.
+ * Parse a non-streaming `chat/completions` JSON body when `stream: true` was ignored.
+ * @returns subjects slice, or `undefined` if the body is not a usable completion object.
  */
-export async function readChatCompletionStreamToSubjects(
-    input: NodeJS.ReadableStream,
+function trySubjectsFromNonStreamCompletionJson(
+    body: string,
     choiceCount: number,
-): Promise<string[]> {
-    if (choiceCount < 1)
-        throw new Error('choiceCount must be at least 1')
+): string[] | undefined {
+    const t = body.trim()
+    if (!t.startsWith('{'))
+        return undefined
+    let json: unknown
+    try {
+        json = JSON.parse(t)
+    }
+    catch {
+        return undefined
+    }
+    if (!json || typeof json !== 'object')
+        return undefined
+    const o = json as { choices?: NonStreamChoice[], error?: { message?: string } }
+    if (o.error)
+        throw new Error(o.error.message || 'OpenAI API error')
+    if (!Array.isArray(o.choices))
+        return undefined
 
     const buffers = Array.from({ length: choiceCount }, () => '')
     let maxIndexSeen = -1
-    const rl = readline.createInterface({ input, crlfDelay: Infinity })
+    for (const ch of o.choices) {
+        const idx = typeof ch.index === 'number' ? ch.index : 0
+        if (idx >= 0 && idx < choiceCount) {
+            buffers[idx] = appendVisibleDelta('', { content: ch.message?.content })
+            maxIndexSeen = Math.max(maxIndexSeen, idx)
+        }
+    }
+    if (maxIndexSeen < 0)
+        return undefined
+    return buffers.slice(0, maxIndexSeen + 1)
+}
 
-    for await (const line of rl) {
+function collectSubjectsFromSseLines(
+    body: string,
+    choiceCount: number,
+): { buffers: string[], maxIndexSeen: number } {
+    const buffers = Array.from({ length: choiceCount }, () => '')
+    let maxIndexSeen = -1
+    for (const line of body.split(/\r?\n/)) {
         const trimmed = line.trim()
         if (!trimmed.startsWith('data:'))
             continue
@@ -93,7 +137,35 @@ export async function readChatCompletionStreamToSubjects(
             throw e
         }
     }
+    return { buffers, maxIndexSeen }
+}
+
+/**
+ * Read an OpenAI-style `chat/completions` response body and return one finished string per choice.
+ * Primary path: SSE lines (`data: {...}`) with `choices[].delta`, bucketed by `choices[].index`
+ * up to `choiceCount` (requested `n`). Returned length is `maxSeenIndex + 1` (capped by `choiceCount`),
+ * mirroring non-stream `choices.length` when fewer parallel completions appear.
+ * Fallback: if no choice index ever appears (e.g. provider ignores `stream: true` and returns one JSON object),
+ * the full body is parsed as a non-streaming completion using `choices[].message.content`.
+ */
+export async function readChatCompletionStreamToSubjects(
+    input: NodeJS.ReadableStream,
+    choiceCount: number,
+): Promise<string[]> {
+    if (choiceCount < 1)
+        throw new Error('choiceCount must be at least 1')
+
+    const body = await readableToUtf8String(input)
+    const { buffers, maxIndexSeen } = collectSubjectsFromSseLines(body, choiceCount)
+
+    if (maxIndexSeen >= 0)
+        return buffers.slice(0, maxIndexSeen + 1)
+
+    const fromJson = trySubjectsFromNonStreamCompletionJson(body, choiceCount)
+    if (fromJson !== undefined)
+        return fromJson
 
-    const effectiveLen = maxIndexSeen < 0 ? 1 : maxIndexSeen + 1
-    return buffers.slice(0, effectiveLen)
+    throw new Error(
+        'Chat completions response had no streamed choice deltas and is not a parseable non-streaming JSON body with choices (or choices were empty).',
+    )
 }