fix(ai): preserve Responses stream cancellation (#113801)

packages/ai/src/providers/openai-responses-shared.test.ts

@@ -1653,6 +1653,26 @@ describe("processResponsesStream", () => {
     expect(output.usage.input).toBe(7);
   });
 
+  it("preserves cancellation when the SDK swallows the abort and ends iteration", async () => {
+    const abort = new AbortController();
+    const output = createAssistantOutput();
+    async function* silentlyAbortedStream() {
+      yield { type: "response.created", response: { id: "resp_aborted" } };
+      abort.abort();
+    }
+
+    await expect(
+      processResponsesStream(
+        silentlyAbortedStream(),
+        output,
+        new AssistantMessageEventStream(),
+        nativeOpenAIModel,
+        { signal: abort.signal },
+      ),
+    ).rejects.toThrow("Request was aborted");
+    expect(output.responseId).toBe("resp_aborted");
+  });
+
   it.each([
     ["omits arguments", undefined],
     ["sends empty arguments", ""],

packages/ai/src/transports/openai-responses-stream-internal.ts

@@ -48,6 +48,7 @@ import {
   type ResponsesThinkingBlock,
   type TextBlockReference,
 } from "./openai-responses-stream-terminal-internal.js";
+import { transportAbortError } from "./transport-stream-shared.js";
 
 type ResponsesConsumedEventType =
   | "error"
@@ -713,6 +714,11 @@ export async function processResponsesStream<TApi extends Api>(
         throw new ResponsesStreamFailure(failure, event.response);
       }
     }
+    // openai-node turns an aborted SSE iterator into normal completion; preserve
+    // the caller's authoritative reason before classifying terminal stream state.
+    if (options?.signal?.aborted) {
+      throw transportAbortError(options.signal);
+    }
     if (streamingToolCalls.hasActive()) {
       throw new Error("Responses stream ended with unresolved tool calls");
     }