Fix async image concurrent results and Chat Completions stream completion

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);

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);

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<string, unknown> | undefined {
+  return value !== null && typeof value === "object" && !Array.isArray(value)
+    ? (value as Record<string, unknown>)
+    : 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<number, PendingToolCall>();
+
+  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<string>();
+      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;
+    },
+  };
+}

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);
+}

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<ResponsesResponse>({
         path: requestPath,

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": [
       {