fix(ai): report interrupted local-model streams instead of empty success (#129794)

* fix(ai): report interrupted local-model streams instead of empty success

* test(ai): await asynchronous managed stream fixtures

packages/ai/src/transports/openai-completions-stream.tool-calls.test.ts

@@ -244,7 +244,41 @@ describe("openai completions stream", () => {
     }
   });
 
-  it("keeps tool calls fail-closed through fetch wrapper when stream ends without [DONE] and without finish_reason", async () => {
+  it.each([
+    { name: "an empty response", delta: undefined, done: false, finishReason: undefined },
+    {
+      name: "a partial visible response",
+      delta: { content: "A partial answer" },
+      done: false,
+      finishReason: undefined,
+    },
+    {
+      name: "an unfinished native tool call",
+      delta: {
+        tool_calls: [
+          {
+            index: 0,
+            id: "call_loopback_nodone",
+            function: { name: "bash", arguments: '{"cmd":"echo no done"}' },
+          },
+        ],
+      },
+      done: false,
+      finishReason: undefined,
+    },
+    {
+      name: "a clean SSE terminal without finish_reason",
+      delta: { content: "A complete answer" },
+      done: true,
+      finishReason: undefined,
+    },
+    {
+      name: "an explicit finish_reason without an SSE terminal",
+      delta: { content: "A complete answer" },
+      done: false,
+      finishReason: "stop" as const,
+    },
+  ])("preserves an honest terminal outcome for $name", async ({ delta, done, finishReason }) => {
     const server = createServer((req, res) => {
       let body = "";
       req.setEncoding("utf8");
@@ -258,21 +292,12 @@ describe("openai completions stream", () => {
           "cache-control": "no-cache",
           connection: "keep-alive",
         });
-        // Emit delta.tool_calls chunk with no finish_reason
-        res.write(
-          `data: ${JSON.stringify(
-            makeCompletionsChunk({
-              tool_calls: [
-                {
-                  index: 0,
-                  id: "call_loopback_nodone",
-                  function: { name: "bash", arguments: '{"cmd":"echo no done"}' },
-                },
-              ],
-            }),
-          )}\n\n`,
-        );
-        // Close WITHOUT data: [DONE] — simulates connection drop / truncated stream
+        if (delta) {
+          res.write(`data: ${JSON.stringify(makeCompletionsChunk(delta, finishReason))}\n\n`);
+        }
+        if (done) {
+          res.write("data: [DONE]\n\n");
+        }
         res.end();
       });
     });
@@ -303,26 +328,23 @@ describe("openai completions stream", () => {
         { apiKey: "test-key" } as never,
       );
 
-      let doneReason: string | undefined;
-      const doneMessage: { content?: Array<{ type?: string }> } = {};
+      let terminalEvent: string | undefined;
       for await (const event of stream as AsyncIterable<{
         type: string;
-        reason?: string;
-        message?: { content?: Array<{ type?: string }> };
       }>) {
-        if (event.type === "done") {
-          doneReason = event.reason;
-          if (event.message) {
-            Object.assign(doneMessage, event.message);
-          }
+        if (event.type === "done" || event.type === "error") {
+          terminalEvent = event.type;
         }
       }
 
-      // EOF without [DONE] → sawStreamDONE stays false → fail-closed
-      expect(doneReason).toBe("stop");
-      const toolCallBlocks =
-        doneMessage.content?.filter((block) => block.type === "toolCall") ?? [];
-      expect(toolCallBlocks).toStrictEqual([]);
+      const result = await (await stream).result();
+      const cleanTerminal = done || Boolean(finishReason);
+      expect(terminalEvent).toBe(cleanTerminal ? "done" : "error");
+      expect(result.stopReason).toBe(cleanTerminal ? "stop" : "error");
+      if (!cleanTerminal) {
+        expect(result.errorMessage).toContain("Stream ended without finish_reason");
+      }
+      expect(result.content.filter((block) => block.type === "toolCall")).toStrictEqual([]);
     } finally {
       await new Promise<void>((resolve, reject) => {
         server.close((error) => (error ? reject(error) : resolve()));

packages/ai/src/transports/openai-completions-stream.ts

@@ -124,7 +124,7 @@ export async function processCompletionsStream(
   const toolCallBlockIndices = new WeakMap<ToolCallBlock, number>();
   let explicitVisibleTextBlocks: Set<TextBlock> | undefined;
   const normalizeToolCallDeltas = createOpenAICompletionsToolCallDeltaNormalizer();
-  let sawStopFinishReason = false;
+  let finishReason: string | undefined;
   let sawNativeToolCallDelta = false;
   const blockIndex = () => output.content.length - 1;
   const measureUtf8Bytes = (text: string) => Buffer.byteLength(text, "utf8");
@@ -442,9 +442,7 @@ export async function processCompletionsStream(
         allowSingularToolCall: true,
       });
       output.stopReason = finishReasonResult.stopReason;
-      if (finishReasonResult.stopReason === "stop") {
-        sawStopFinishReason = true;
-      }
+      finishReason = finishReasonResult.stopReason;
       if (finishReasonResult.errorMessage) {
         output.errorMessage = finishReasonResult.errorMessage;
       }
@@ -567,22 +565,18 @@ export async function processCompletionsStream(
     emitReasoningUsageActivity(hasReasoningUsageActivity);
     await cooperativeScheduler.afterEvent();
   }
+  if (!finishReason && options?.sawStreamDONE?.() === false) {
+    throw new Error("Stream ended without finish_reason");
+  }
   flushReasoningTagTextPartitioner();
   flushDeepSeekToolCallRecovererAtEnd();
   flushDeepSeekTextFilterAtEnd();
   currentBlock = null;
   flushPendingPostToolCallDeltas();
-  // Promote complete silent tool-call-only responses when the stream finished
-  // cleanly (reached post-loop). Two paths:
-  //   sawStopFinishReason: explicit provider terminal (legacy DSML / #88791)
-  //   sawNativeToolCallDelta + sawStreamDONE: structured delta.tool_calls with
-  //     a clean SSE [DONE] terminal but no finish_reason (e.g. Evolink
-  //     DeepSeek V4). [DONE] tracking distinguishes clean termination from
-  //     connection drops (EOF without [DONE] remains fail-closed).
-  // Truncated streams throw before reaching this code.
+  // Only an explicit stop or observed SSE terminal may authorize silent tool calls.
   finalizeOpenAICompletionsToolCalls(output, {
     allowSilentToolCallPromotion:
-      sawStopFinishReason || (sawNativeToolCallDelta && (options?.sawStreamDONE?.() ?? false)),
+      finishReason === "stop" || (sawNativeToolCallDelta && (options?.sawStreamDONE?.() ?? false)),
     onConfirmedToolCall(block, contentIndex) {
       if (block.type !== "toolCall") {
         return;