From a2956d5f9a138af3a77a8747efacf3a7b8673ff2 Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Mon, 2 Nov 2026 15:16:43 +0800 Subject: [PATCH 1/2] fix(image): collect every async concurrent result image Prefer choices per task, then fall back to results, so concurrent waits no longer keep only the first non-empty task. --- packages/commands/src/commands/image/edit.ts | 19 +-- .../commands/src/commands/image/generate.ts | 19 +-- .../src/commands/shared/task-images.ts | 11 ++ .../tests/output/image-output.test.ts | 160 ++++++++++++++++-- 4 files changed, 158 insertions(+), 51 deletions(-) create mode 100644 packages/commands/src/commands/shared/task-images.ts diff --git a/packages/commands/src/commands/image/edit.ts b/packages/commands/src/commands/image/edit.ts index 8a38b27d..6fe9aa8c 100644 --- a/packages/commands/src/commands/image/edit.ts +++ b/packages/commands/src/commands/image/edit.ts @@ -42,6 +42,7 @@ import { mcpSingleImageOnlyError, type Localize, } from "../shared/aigc-mcp.ts"; +import { collectTaskImageUrls } from "../shared/task-images.ts"; const EDIT_FLAGS = { image: { @@ -491,23 +492,7 @@ async function handleAsyncMode( ); const results = await Promise.all(pollPromises); - - let imageUrls: string[] = []; - for (const result of results) { - if (result.output.choices) { - const urls = result.output.choices - .flatMap((choice) => choice.message?.content || []) - .map((item) => item.image) - .filter(Boolean); - imageUrls.push(...urls); - } - if (result.output.results) { - const urls = result.output.results.map((item) => item.url).filter(Boolean); - if (urls.length > 0 && imageUrls.length === 0) { - imageUrls.push(...urls); - } - } - } + const imageUrls = results.flatMap((result) => collectTaskImageUrls(result.output)); if (imageUrls.length === 0) { throw new BailianError("All tasks completed but no images returned.", ExitCode.GENERAL); diff --git a/packages/commands/src/commands/image/generate.ts b/packages/commands/src/commands/image/generate.ts index 0997c072..caf0f82d 100644 --- a/packages/commands/src/commands/image/generate.ts +++ b/packages/commands/src/commands/image/generate.ts @@ -38,6 +38,7 @@ import { withAigcMcpSession, type Localize, } from "../shared/aigc-mcp.ts"; +import { collectTaskImageUrls } from "../shared/task-images.ts"; import { join } from "path"; @@ -394,23 +395,7 @@ async function handleAsyncMode( }); const results = await Promise.all(pollPromises); - - let imageUrls: string[] = []; - for (const result of results) { - if (result.output.choices) { - const urls = result.output.choices - .flatMap((choice) => choice.message?.content || []) - .map((item) => item.image) - .filter(Boolean); - imageUrls.push(...urls); - } - if (result.output.results) { - const urls = result.output.results.map((item) => item.url).filter(Boolean); - if (urls.length > 0 && imageUrls.length === 0) { - imageUrls.push(...urls); - } - } - } + const imageUrls = results.flatMap((result) => collectTaskImageUrls(result.output)); if (imageUrls.length === 0) { throw new BailianError("All tasks completed but no images returned.", ExitCode.GENERAL); diff --git a/packages/commands/src/commands/shared/task-images.ts b/packages/commands/src/commands/shared/task-images.ts new file mode 100644 index 00000000..05f92422 --- /dev/null +++ b/packages/commands/src/commands/shared/task-images.ts @@ -0,0 +1,11 @@ +import type { DashScopeTaskResponse } from "bailian-cli-core"; + +/** Image URLs from one async task: use choices when present, otherwise results. */ +export function collectTaskImageUrls(output: DashScopeTaskResponse["output"]): string[] { + const fromChoices = (output.choices ?? []) + .flatMap((choice) => choice.message?.content ?? []) + .map((item) => item.image) + .filter(Boolean); + if (fromChoices.length > 0) return fromChoices; + return (output.results ?? []).map((item) => item.url).filter(Boolean); +} diff --git a/packages/commands/tests/output/image-output.test.ts b/packages/commands/tests/output/image-output.test.ts index b60ef1df..d316e651 100644 --- a/packages/commands/tests/output/image-output.test.ts +++ b/packages/commands/tests/output/image-output.test.ts @@ -15,6 +15,7 @@ let origin: string; let scenario: "success" | "submit-error" | "task-failed" | "empty" | "download-error"; let requests: Array<{ method: string; path: string; body: unknown; async: boolean }>; let legacyResults: boolean; +let asyncImageShape: "default" | "results" | "mixed" | "both" | "partial-empty" | "all-empty"; let submissions: number; let downloads: number; @@ -25,6 +26,7 @@ beforeEach(async () => { submissions = 0; downloads = 0; legacyResults = false; + asyncImageShape = "default"; server = createServer(async (request, response) => { if (request.url?.startsWith("/files/")) { downloads++; @@ -74,6 +76,13 @@ beforeEach(async () => { return; } const taskId = request.url!.split("/").at(-1)!; + const taskNumber = taskId.slice(5); + const fileUrl = `${origin}/files/${taskNumber}.png`; + const choiceImage = { message: { content: [{ image: fileUrl }] } }; + const resultImage = { url: fileUrl }; + const ignoredImage = { url: `${origin}/files/ignored-${taskNumber}.png` }; + const useResults = legacyResults || asyncImageShape === "results"; + const images = taskImages(taskNumber, choiceImage, resultImage, ignoredImage, useResults); response.end( JSON.stringify({ output: { @@ -81,23 +90,7 @@ beforeEach(async () => { task_status: scenario === "task-failed" ? "FAILED" : "SUCCEEDED", ...(scenario === "task-failed" ? { code: "TaskFailure", message: "Original task failure" } - : legacyResults - ? { - results: - scenario === "empty" ? [] : [{ url: `${origin}/files/${taskId.slice(5)}.png` }], - } - : { - choices: - scenario === "empty" - ? [] - : [ - { - message: { - content: [{ image: `${origin}/files/${taskId.slice(5)}.png` }], - }, - }, - ], - }), + : images), }, }), ); @@ -117,6 +110,25 @@ afterEach(async () => { rmSync(directory, { recursive: true, force: true }); }); +function taskImages( + taskNumber: string, + choiceImage: { message: { content: Array<{ image: string }> } }, + resultImage: { url: string }, + ignoredImage: { url: string }, + useResults: boolean, +) { + if (asyncImageShape === "all-empty") return { results: [] }; + if (asyncImageShape === "partial-empty") { + return taskNumber === "2" ? { results: [] } : { results: [resultImage] }; + } + if (asyncImageShape === "both") return { choices: [choiceImage], results: [ignoredImage] }; + if (asyncImageShape === "mixed") { + return taskNumber === "1" ? { choices: [choiceImage] } : { results: [resultImage] }; + } + if (useResults) return { results: scenario === "empty" ? [] : [resultImage] }; + return { choices: scenario === "empty" ? [] : [choiceImage] }; +} + const commands = [ { name: "generate", command: generate }, { name: "edit", command: edit }, @@ -381,3 +393,117 @@ for (const entry of commands) { expect(result.stdout).not.toContain("cGF5bG9hZA=="); }); } + +function savedPaths(count: number) { + const paths: string[] = []; + for (let index = 1; index <= count; index += 1) { + paths.push(join(directory, "explicit", `fixed_${String(index).padStart(3, "0")}.png`)); + } + return paths; +} + +function fileUrls(numbers: number[]) { + return numbers.map((number) => `${origin}/files/${number}.png`); +} + +for (const entry of commands) { + test(`${entry.name} async concurrent results keeps every task image`, async () => { + asyncImageShape = "results"; + const snapshots = []; + for (const format of ["json", "text"] as const) { + submissions = 0; + downloads = 0; + const result = await invoke(entry.command, flagsFor(entry.name, "wait", 3), { + format, + quiet: true, + }); + expect(result.exitCode, result.stderr).toBe(0); + const data = format === "json" ? JSON.parse(result.stdout) : parseYaml(result.stdout); + const saved = savedPaths(3); + expect(data).toEqual({ + urls: fileUrls([1, 2, 3]), + saved, + total: 3, + ...(entry.name === "generate" ? { task_ids: ["task-1", "task-2", "task-3"] } : {}), + }); + for (const path of saved) expect(readFileSync(path, "utf8")).toBe("payload"); + expect(downloads).toBe(3); + snapshots.push(data); + } + expect(snapshots[1]).toEqual(snapshots[0]); + }); + + test(`${entry.name} async mixed choices and results keeps both tasks`, async () => { + asyncImageShape = "mixed"; + const result = await invoke(entry.command, flagsFor(entry.name, "wait", 2), { + format: "json", + quiet: true, + }); + expect(result.exitCode, result.stderr).toBe(0); + expect(JSON.parse(result.stdout)).toEqual({ + urls: fileUrls([1, 2]), + saved: savedPaths(2), + total: 2, + ...(entry.name === "generate" ? { task_ids: ["task-1", "task-2"] } : {}), + }); + expect(downloads).toBe(2); + }); + + test(`${entry.name} async task with both fields keeps choices only`, async () => { + asyncImageShape = "both"; + const result = await invoke(entry.command, flagsFor(entry.name, "wait", 2), { + format: "json", + quiet: true, + }); + expect(result.exitCode, result.stderr).toBe(0); + const data = JSON.parse(result.stdout); + expect(data.urls).toEqual(fileUrls([1, 2])); + expect(data.total).toBe(2); + expect(data.saved).toEqual(savedPaths(2)); + expect(JSON.stringify(data)).not.toContain("ignored-"); + expect(downloads).toBe(2); + }); + + test(`${entry.name} async partial empty results keeps the remaining images`, async () => { + asyncImageShape = "partial-empty"; + const result = await invoke(entry.command, flagsFor(entry.name, "wait", 3), { + format: "json", + quiet: true, + }); + expect(result.exitCode, result.stderr).toBe(0); + expect(JSON.parse(result.stdout)).toEqual({ + urls: fileUrls([1, 3]), + saved: savedPaths(2), + total: 2, + ...(entry.name === "generate" ? { task_ids: ["task-1", "task-2", "task-3"] } : {}), + }); + expect(downloads).toBe(2); + }); + + test(`${entry.name} async empty results still fails without images`, async () => { + asyncImageShape = "all-empty"; + const result = await invoke(entry.command, flagsFor(entry.name, "wait", 2), { + format: "json", + quiet: true, + }); + expect(result.exitCode).toBe(1); + expect(result.stdout).toBe(""); + expect(result.stderr).toContain("All tasks completed but no images returned."); + expect(downloads).toBe(0); + }); + + test(`${entry.name} async concurrent choices still collects every image`, async () => { + const result = await invoke(entry.command, flagsFor(entry.name, "wait", 3), { + format: "json", + quiet: true, + }); + expect(result.exitCode, result.stderr).toBe(0); + expect(JSON.parse(result.stdout)).toEqual({ + urls: fileUrls([1, 2, 3]), + saved: savedPaths(3), + total: 3, + ...(entry.name === "generate" ? { task_ids: ["task-1", "task-2", "task-3"] } : {}), + }); + expect(downloads).toBe(3); + }); +} From b5775f0ad33ab73e7a918d663593ef7ff373dc1d Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Mon, 2 Nov 2026 15:43:27 +0800 Subject: [PATCH 2/2] fix(text-chat): validate Chat Completions stream completion and aggregate tool_calls --- .../commands/shared/chat-completion-stream.ts | 182 +++++++++ packages/commands/src/commands/text/chat.ts | 48 +-- .../output/fixtures/inference-output.json | 15 +- .../commands/tests/text-chat-stream.test.ts | 380 ++++++++++++++++++ 4 files changed, 595 insertions(+), 30 deletions(-) create mode 100644 packages/commands/src/commands/shared/chat-completion-stream.ts create mode 100644 packages/commands/tests/text-chat-stream.test.ts diff --git a/packages/commands/src/commands/shared/chat-completion-stream.ts b/packages/commands/src/commands/shared/chat-completion-stream.ts new file mode 100644 index 00000000..16178bab --- /dev/null +++ b/packages/commands/src/commands/shared/chat-completion-stream.ts @@ -0,0 +1,182 @@ +import { BailianError, ExitCode, type LocalizedText } from "bailian-cli-core"; +import { parseModelStreamEvent } from "./model-stream.ts"; + +export interface ChatStreamUpdate { + content: string; + reasoning: string; +} + +export interface ChatStreamToolCall { + id: string; + type: "function"; + function: { name: string; arguments: string }; +} + +export interface ChatStreamResult { + content: string; + tool_calls?: ChatStreamToolCall[]; + finish_reason: string | null; +} + +interface PendingToolCall { + id: string; + name: string; + arguments: string; +} + +function asRecord(value: unknown): Record | undefined { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +/** Aggregate Chat Completions SSE deltas with completion and tool-call checks. */ +export function createChatCompletionStreamAccumulator(localize: (text: LocalizedText) => string): { + readonly completed: boolean; + accept(event: { event?: string; data: string }): ChatStreamUpdate; + finish(): ChatStreamResult; +} { + let completed = false; + let content = ""; + let finishReason: string | null = null; + const toolCalls = new Map(); + + const missingIndexError = () => + new BailianError( + localize({ + "en-US": "Tool call delta is missing a valid index.", + "zh-CN": "工具调用增量缺少有效的 index。", + }), + ExitCode.GENERAL, + ); + + const conflictingIdError = () => + new BailianError( + localize({ + "en-US": "Tool call index returned conflicting ids.", + "zh-CN": "同一工具调用 index 返回了不同的 id。", + }), + ExitCode.GENERAL, + ); + + const acceptToolCallDelta = (fragment: unknown): void => { + const delta = asRecord(fragment); + if (!delta) throw missingIndexError(); + + const index = delta.index; + if (typeof index !== "number" || !Number.isInteger(index) || index < 0) { + throw missingIndexError(); + } + + const entry = toolCalls.get(index) ?? { id: "", name: "", arguments: "" }; + if (typeof delta.id === "string" && delta.id) { + if (entry.id && entry.id !== delta.id) throw conflictingIdError(); + entry.id = delta.id; + } + + const functionDelta = asRecord(delta.function); + if (typeof functionDelta?.name === "string") entry.name += functionDelta.name; + if (typeof functionDelta?.arguments === "string") entry.arguments += functionDelta.arguments; + + toolCalls.set(index, entry); + }; + + return { + get completed() { + return completed; + }, + + accept(event: { event?: string; data: string }): ChatStreamUpdate { + let contentDelta = ""; + let reasoningDelta = ""; + + if (event.data === "[DONE]") { + completed = true; + return { content: "", reasoning: "" }; + } + + const parsed = parseModelStreamEvent(event); + if (!parsed) return { content: "", reasoning: "" }; + + const choices = Array.isArray(parsed.choices) ? parsed.choices : []; + for (const choiceValue of choices) { + const choice = asRecord(choiceValue); + if (!choice) continue; + + const delta = asRecord(choice.delta); + if (delta) { + if (typeof delta.reasoning_content === "string" && delta.reasoning_content) { + reasoningDelta += delta.reasoning_content; + } + if (typeof delta.content === "string" && delta.content) { + contentDelta += delta.content; + content += delta.content; + } + if (Array.isArray(delta.tool_calls)) { + for (const fragment of delta.tool_calls) acceptToolCallDelta(fragment); + } + } + + if (typeof choice.finish_reason === "string" && choice.finish_reason) { + if (finishReason === null) finishReason = choice.finish_reason; + completed = true; + } + } + + return { content: contentDelta, reasoning: reasoningDelta }; + }, + + finish(): ChatStreamResult { + if (!completed) { + throw new BailianError( + localize({ + "en-US": + "Stream disconnected before completion: stream closed before [DONE] or finish_reason.", + "zh-CN": "流在完成前断开：未收到 [DONE] 或 finish_reason。", + }), + ExitCode.GENERAL, + ); + } + + const sorted = [...toolCalls.entries()] + .sort(([left], [right]) => left - right) + .map(([, call]) => call); + + const seenIds = new Set(); + const tool_calls: ChatStreamToolCall[] = []; + for (const call of sorted) { + if (!call.id || !call.name) { + throw new BailianError( + localize({ + "en-US": "Tool call is missing a required id or function name.", + "zh-CN": "工具调用缺少必需的 id 或 function.name。", + }), + ExitCode.GENERAL, + ); + } + if (seenIds.has(call.id)) { + throw new BailianError( + localize({ + "en-US": "Duplicate tool call id across indexes.", + "zh-CN": "不同工具调用 index 返回了重复的 id。", + }), + ExitCode.GENERAL, + ); + } + seenIds.add(call.id); + tool_calls.push({ + id: call.id, + type: "function", + function: { name: call.name, arguments: call.arguments }, + }); + } + + const result: ChatStreamResult = { + content, + finish_reason: finishReason, + }; + if (tool_calls.length > 0) result.tool_calls = tool_calls; + return result; + }, + }; +} diff --git a/packages/commands/src/commands/text/chat.ts b/packages/commands/src/commands/text/chat.ts index 83627491..d4937c9c 100644 --- a/packages/commands/src/commands/text/chat.ts +++ b/packages/commands/src/commands/text/chat.ts @@ -1,4 +1,4 @@ -import { parseModelStreamEvent } from "../shared/model-stream.ts"; +import { createChatCompletionStreamAccumulator } from "../shared/chat-completion-stream.ts"; import { defineCommand, resolveBatchChatEndpoint, @@ -13,7 +13,6 @@ import { type ResponsesRequest, type ResponsesResponse, type ResponsesStreamEvent, - type StreamChunk, type FlagsDef, type ParsedFlags, } from "bailian-cli-core"; @@ -338,6 +337,7 @@ export default defineCommand({ let textContent = ""; let inThinking = false; let responsesCompleted = false; + const accumulator = createChatCompletionStreamAccumulator(ctx.localize); const writesStreamingStdout = format === "text"; const isTTY = process.stdout.isTTY; const statusOut = @@ -346,8 +346,8 @@ export default defineCommand({ const statusColor = ansi(statusOut); for await (const event of parseSSE(responseStream)) { - if (event.data === "[DONE]") break; if (api === "responses") { + if (event.data === "[DONE]") break; let parsedEvent: ResponsesStreamEvent; try { parsedEvent = JSON.parse(event.data) as ResponsesStreamEvent; @@ -367,36 +367,36 @@ export default defineCommand({ continue; } - const parsed = parseModelStreamEvent(event) as unknown as StreamChunk | undefined; - if (!parsed) continue; + const update = accumulator.accept(event); - for (const choice of parsed.choices ?? []) { - const delta = choice.delta; - - // Handle thinking/reasoning content - if (delta.reasoning_content) { - if (writesStreamingStdout && !inThinking) { - inThinking = true; - statusOut.write(statusColor.dim("Thinking:\n")); - } - if (writesStreamingStdout) statusOut.write(delta.reasoning_content); + // Write thinking/reasoning content to the status stream. + if (update.reasoning) { + if (writesStreamingStdout && !inThinking) { + inThinking = true; + statusOut.write(statusColor.dim("Thinking:\n")); } + if (writesStreamingStdout) statusOut.write(update.reasoning); + } - // Handle regular content - if (delta.content) { - if (writesStreamingStdout && inThinking) { - statusOut.write(`${statusColor.reset}\n\nResponse:\n`); - inThinking = false; - } - textContent += delta.content; - if (writesStreamingStdout) resultOut.write(delta.content); + // Write regular content to stdout; tool_calls stay off the text stream. + if (update.content) { + if (writesStreamingStdout && inThinking) { + statusOut.write(`${statusColor.reset}\n\nResponse:\n`); + inThinking = false; } + if (writesStreamingStdout) resultOut.write(update.content); } + + if (accumulator.completed) break; } if (api === "responses") assertResponsesStreamCompleted(responsesCompleted); if (inThinking) statusOut.write(statusColor.reset); - ctx.output.result({ content: textContent }, format === "text" ? { text: "" } : undefined); + if (api === "responses") { + ctx.output.result({ content: textContent }, format === "text" ? { text: "" } : undefined); + } else { + ctx.output.result(accumulator.finish(), format === "text" ? { text: "" } : undefined); + } } else if (api === "responses") { const response = await ctx.client.requestJson({ path: requestPath, diff --git a/packages/commands/tests/output/fixtures/inference-output.json b/packages/commands/tests/output/fixtures/inference-output.json index d620fabf..ffbfe234 100644 --- a/packages/commands/tests/output/fixtures/inference-output.json +++ b/packages/commands/tests/output/fixtures/inference-output.json @@ -83,9 +83,10 @@ ] }, "text chat:json:true": { - "stdout": "{\n \"content\": \"Hello fixture\"\n}\n", + "stdout": "{\n \"content\": \"Hello fixture\",\n \"finish_reason\": \"stop\"\n}\n", "data": { - "content": "Hello fixture" + "content": "Hello fixture", + "finish_reason": "stop" }, "requests": [ { @@ -126,9 +127,10 @@ ] }, "text stream:json:false": { - "stdout": "{\n \"content\": \"Hello fixture\"\n}\n", + "stdout": "{\n \"content\": \"Hello fixture\",\n \"finish_reason\": \"stop\"\n}\n", "data": { - "content": "Hello fixture" + "content": "Hello fixture", + "finish_reason": "stop" }, "requests": [ { @@ -169,9 +171,10 @@ ] }, "text stream:json:true": { - "stdout": "{\n \"content\": \"Hello fixture\"\n}\n", + "stdout": "{\n \"content\": \"Hello fixture\",\n \"finish_reason\": \"stop\"\n}\n", "data": { - "content": "Hello fixture" + "content": "Hello fixture", + "finish_reason": "stop" }, "requests": [ { diff --git a/packages/commands/tests/text-chat-stream.test.ts b/packages/commands/tests/text-chat-stream.test.ts new file mode 100644 index 00000000..ae7755aa --- /dev/null +++ b/packages/commands/tests/text-chat-stream.test.ts @@ -0,0 +1,380 @@ +import { buildSettings, BailianError, ExitCode } from "bailian-cli-core"; +import { afterEach, describe, expect, test, vi } from "vite-plus/test"; +import chatCommand from "../src/commands/text/chat.ts"; +import { withCommandOutput } from "./output/command-context.ts"; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +function chatStreamResponse(events: Array>): Response { + const body = events + .map((event) => { + if (typeof event === "string") { + if (event.startsWith("event:")) return `${event}\n\n`; + return `data: ${event}\n\n`; + } + return `data: ${JSON.stringify(event)}\n\n`; + }) + .join(""); + return new Response(body, { headers: { "content-type": "text/event-stream" } }); +} + +function chunk( + delta: Record, + finishReason: string | null = null, +): Record { + return { + id: "chatcmpl-fixture", + object: "chat.completion.chunk", + choices: [{ index: 0, delta, finish_reason: finishReason }], + }; +} + +async function runChatStream(options: { + events: Array>; + output?: "json" | "text"; + captureStdout?: boolean; +}): Promise<{ error?: BailianError; stdout: string }> { + let stdout = ""; + if (options.captureStdout !== false) { + vi.spyOn(process.stdout, "write").mockImplementation((chunkValue) => { + stdout += typeof chunkValue === "string" ? chunkValue : String(chunkValue); + return true; + }); + } else { + vi.spyOn(process.stdout, "write").mockImplementation(() => true); + } + + try { + await chatCommand.run( + withCommandOutput({ + settings: buildSettings({ + flags: { output: options.output ?? "json" }, + env: {}, + file: {}, + }), + flags: { + api: "chat", + message: ["hello"], + stream: true, + }, + client: { + request: async () => chatStreamResponse(options.events), + }, + }) as never, + ); + return { stdout }; + } catch (error) { + expect(error).toBeInstanceOf(BailianError); + return { error: error as BailianError, stdout }; + } +} + +describe("Chat Completions stream", () => { + test("normal stop with finish_reason and [DONE] succeeds", async () => { + const { error, stdout } = await runChatStream({ + events: [chunk({ content: "Hello " }), chunk({ content: "world" }, "stop"), "[DONE]"], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout)).toEqual({ + content: "Hello world", + finish_reason: "stop", + }); + }); + + test("finish_reason alone without [DONE] succeeds", async () => { + const { error, stdout } = await runChatStream({ + events: [chunk({ content: "ok" }, "stop")], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout)).toEqual({ + content: "ok", + finish_reason: "stop", + }); + }); + + test("[DONE] alone succeeds with null finish_reason", async () => { + const { error, stdout } = await runChatStream({ + events: [chunk({ content: "done-only" }), "[DONE]"], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout)).toEqual({ + content: "done-only", + finish_reason: null, + }); + }); + + test("empty completion with only [DONE] succeeds", async () => { + const { error, stdout } = await runChatStream({ + events: ["[DONE]"], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout)).toEqual({ + content: "", + finish_reason: null, + }); + }); + + test("early EOF before completion throws GENERAL", async () => { + const { error, stdout } = await runChatStream({ + events: [chunk({ content: "partial" })], + }); + + expect(error?.exitCode).toBe(ExitCode.GENERAL); + expect(error?.message).toBe( + "Stream disconnected before completion: stream closed before [DONE] or finish_reason.", + ); + expect(stdout).toBe(""); + }); + + test("finish_reason length succeeds without error", async () => { + const { error, stdout } = await runChatStream({ + events: [chunk({ content: "cut" }, "length"), "[DONE]"], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout)).toEqual({ + content: "cut", + finish_reason: "length", + }); + }); + + test("tool-only response returns structured tool_calls", async () => { + const { error, stdout } = await runChatStream({ + events: [ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: '{"city":"北京"}' }, + }, + ], + }), + chunk({}, "tool_calls"), + "[DONE]", + ], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout)).toEqual({ + content: "", + tool_calls: [ + { + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: '{"city":"北京"}' }, + }, + ], + finish_reason: "tool_calls", + }); + }); + + test("multi-fragment tool arguments concatenate in order", async () => { + const { error, stdout } = await runChatStream({ + events: [ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: '{"city":"' }, + }, + ], + }), + chunk({ + tool_calls: [{ index: 0, function: { arguments: '北京"}' } }], + }), + chunk({}, "tool_calls"), + "[DONE]", + ], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout).tool_calls[0].function.arguments).toBe('{"city":"北京"}'); + }); + + test("interleaved tool indexes sort ascending without mixing args", async () => { + const { error, stdout } = await runChatStream({ + events: [ + chunk({ + tool_calls: [ + { + index: 1, + id: "call_b", + type: "function", + function: { name: "second", arguments: '{"b":' }, + }, + ], + }), + chunk({ + tool_calls: [ + { + index: 0, + id: "call_a", + type: "function", + function: { name: "first", arguments: '{"a":1}' }, + }, + ], + }), + chunk({ + tool_calls: [{ index: 1, function: { arguments: "2}" } }], + }), + chunk({}, "tool_calls"), + "[DONE]", + ], + }); + + expect(error).toBeUndefined(); + expect(JSON.parse(stdout).tool_calls).toEqual([ + { + id: "call_a", + type: "function", + function: { name: "first", arguments: '{"a":1}' }, + }, + { + id: "call_b", + type: "function", + function: { name: "second", arguments: '{"b":2}' }, + }, + ]); + }); + + test("conflicting ids on the same index throw a protocol error", async () => { + const { error } = await runChatStream({ + events: [ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: "{" }, + }, + ], + }), + chunk({ + tool_calls: [{ index: 0, id: "call_2", function: { arguments: "}" } }], + }), + ], + }); + + expect(error?.exitCode).toBe(ExitCode.GENERAL); + expect(error?.message).toBe("Tool call index returned conflicting ids."); + }); + + test("duplicate ids across indexes throw a protocol error", async () => { + const { error } = await runChatStream({ + events: [ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_same", + type: "function", + function: { name: "first", arguments: "{}" }, + }, + { + index: 1, + id: "call_same", + type: "function", + function: { name: "second", arguments: "{}" }, + }, + ], + }), + chunk({}, "tool_calls"), + "[DONE]", + ], + }); + + expect(error?.exitCode).toBe(ExitCode.GENERAL); + expect(error?.message).toBe("Duplicate tool call id across indexes."); + }); + + test("finish rejects tool calls missing id or name", async () => { + const { error } = await runChatStream({ + events: [ + chunk({ + tool_calls: [{ index: 0, function: { arguments: '{"city":"北京"}' } }], + }), + chunk({}, "tool_calls"), + "[DONE]", + ], + }); + + expect(error?.exitCode).toBe(ExitCode.GENERAL); + expect(error?.message).toBe("Tool call is missing a required id or function name."); + }); + + test("tool delta missing index throws a protocol error", async () => { + const { error } = await runChatStream({ + events: [ + chunk({ + tool_calls: [ + { + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: "{}" }, + }, + ], + }), + ], + }); + + expect(error?.exitCode).toBe(ExitCode.GENERAL); + expect(error?.message).toBe("Tool call delta is missing a valid index."); + }); + + test("service error event preserves the original message", async () => { + const { error } = await runChatStream({ + events: [ + `event: error\ndata: ${JSON.stringify({ message: "provider quota exceeded", code: "quota" })}`, + ], + }); + + expect(error?.exitCode).toBe(ExitCode.GENERAL); + expect(error?.message).toBe("provider quota exceeded"); + }); + + test("text mode streams content once without reprinting the final result", async () => { + const { error, stdout } = await runChatStream({ + output: "text", + events: [chunk({ content: "Hello " }), chunk({ content: "world" }, "stop"), "[DONE]"], + }); + + expect(error).toBeUndefined(); + // Reporter still ends the empty text result with a trailing newline. + expect(stdout).toBe("Hello world\n"); + }); + + test("text mode tool-only response does not render tool_calls", async () => { + const { error, stdout } = await runChatStream({ + output: "text", + events: [ + chunk({ + tool_calls: [ + { + index: 0, + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: '{"city":"北京"}' }, + }, + ], + }), + chunk({}, "tool_calls"), + "[DONE]", + ], + }); + + expect(error).toBeUndefined(); + // Empty text result still emits the reporter's trailing newline only. + expect(stdout).toBe("\n"); + }); +});