fix(ai): reject non-streaming chat on run errors

.changeset/calm-lions-throw.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai': patch
+---
+
+Reject non-streaming text calls when their event stream ends with a `RUN_ERROR`.

packages/ai/src/activities/chat/stream/processor.ts

@@ -22,8 +22,9 @@ import {
   generateMessageId,
   uiMessageToModelMessages,
 } from '../messages.js'
-import { normalizeToolResult } from '../../../utilities/tool-result'
+import { runErrorEventToError } from '../../../utilities/errors'
 import { isProviderExecutedToolCall } from '../../../utilities/provider-executed'
+import { normalizeToolResult } from '../../../utilities/tool-result'
 import { defaultJSONParser } from './json-parser'
 import {
   appendStructuredOutputDelta,
@@ -1681,15 +1682,7 @@ export class StreamProcessor {
     // the surfaced Error so consumers can recover the upstream detail that the
     // RUN_ERROR's `message` alone discards. Both are optional and added only
     // when present, keeping the Error backward compatible.
-    const error = new Error(errorMessage)
-    const code = chunk.code ?? chunk.error?.code
-    if (code !== undefined) {
-      Object.assign(error, { code })
-    }
-    if (chunk.rawEvent !== undefined) {
-      Object.assign(error, { rawEvent: chunk.rawEvent })
-    }
-    this.events.onError?.(error)
+    this.events.onError?.(runErrorEventToError(chunk))
   }
 
   /**

packages/ai/src/stream-to-response.ts

@@ -9,6 +9,7 @@ import { notifyRunDisconnected } from './delivery-disconnect'
 import { resolveResumeRunId } from './stream-durability'
 import { EventType } from './types'
 import { resolveDebugOption } from './logger/resolve'
+import { runErrorEventToError } from './utilities/errors'
 import type { LockStore } from './activities/chat/middleware/locks'
 import type {
   RunRecord,
@@ -46,6 +47,10 @@ export async function streamToText(
   let accumulatedContent = ''
 
   for await (const chunk of stream) {
+    if (chunk.type === 'RUN_ERROR') {
+      throw runErrorEventToError(chunk)
+    }
+
     if (chunk.type === 'TEXT_MESSAGE_CONTENT' && chunk.delta) {
       accumulatedContent += chunk.delta
     }

packages/ai/src/utilities/errors.ts

@@ -1,3 +1,5 @@
+import type { StreamChunk } from '../types'
+
 /**
  * Best-effort extraction of a human-readable message from an unknown thrown
  * value, returning `undefined` when none can be found.
@@ -27,3 +29,24 @@ export function errorTypeName(err: unknown): string {
   }
   return 'Error'
 }
+
+/**
+ * Convert an AG-UI RUN_ERROR event to the Error shape exposed to consumers.
+ * Preserves the provider code and sanitized raw event when available, while
+ * accepting the deprecated nested error payload for backward compatibility.
+ */
+export function runErrorEventToError(
+  chunk: Extract<StreamChunk, { type: 'RUN_ERROR' }>,
+): Error {
+  const error = new Error(
+    chunk.message || chunk.error?.message || 'An error occurred',
+  )
+  const code = chunk.code ?? chunk.error?.code
+  if (code !== undefined) {
+    Object.assign(error, { code })
+  }
+  if (chunk.rawEvent !== undefined) {
+    Object.assign(error, { rawEvent: chunk.rawEvent })
+  }
+  return error
+}

packages/ai/tests/stream-to-response.test.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, vi } from 'vitest'
 import {
+  streamToText,
   toServerSentEventsStream,
   toServerSentEventsResponse,
 } from '../src/stream-to-response'
@@ -34,6 +35,36 @@ async function readStream(stream: ReadableStream<Uint8Array>): Promise<string> {
   return result
 }
 
+describe('streamToText', () => {
+  it('rejects on RUN_ERROR instead of returning accumulated text', async () => {
+    const rawEvent = {
+      provider_name: 'test-provider',
+      raw: { reason: 'upstream overloaded' },
+    }
+    const stream = createMockStream([
+      {
+        type: 'TEXT_MESSAGE_CONTENT',
+        messageId: 'msg-1',
+        timestamp: Date.now(),
+        delta: 'partial text',
+      },
+      {
+        type: 'RUN_ERROR',
+        timestamp: Date.now(),
+        message: 'Provider request failed',
+        code: 'rate_limit_exceeded',
+        rawEvent,
+      },
+    ])
+
+    await expect(streamToText(stream)).rejects.toMatchObject({
+      message: 'Provider request failed',
+      code: 'rate_limit_exceeded',
+      rawEvent,
+    })
+  })
+})
+
 describe('toServerSentEventsStream', () => {
   it('should convert chunks to SSE format', async () => {
     const chunks: Array<Record<string, unknown>> = [

testing/e2e/src/routeTree.gen.ts

@@ -49,6 +49,7 @@ import { Route as ApiOpenrouterCostRouteImport } from './routes/api.openrouter-c
 import { Route as ApiOpenaiUsageDetailsRouteImport } from './routes/api.openai-usage-details'
 import { Route as ApiOpenaiShellSkillsWireRouteImport } from './routes/api.openai-shell-skills-wire'
 import { Route as ApiOpenaiCompletedResponseTextRouteImport } from './routes/api.openai-completed-response-text'
+import { Route as ApiNonStreamingRunErrorRouteImport } from './routes/api.non-streaming-run-error'
 import { Route as ApiMultimodalToolResultWireRouteImport } from './routes/api.multimodal-tool-result-wire'
 import { Route as ApiMiddlewareTestRouteImport } from './routes/api.middleware-test'
 import { Route as ApiMessageIdsRouteImport } from './routes/api.message-ids'
@@ -298,6 +299,11 @@ const ApiOpenaiCompletedResponseTextRoute =
     path: '/api/openai-completed-response-text',
     getParentRoute: () => rootRouteImport,
   } as any)
+const ApiNonStreamingRunErrorRoute = ApiNonStreamingRunErrorRouteImport.update({
+  id: '/api/non-streaming-run-error',
+  path: '/api/non-streaming-run-error',
+  getParentRoute: () => rootRouteImport,
+} as any)
 const ApiMultimodalToolResultWireRoute =
   ApiMultimodalToolResultWireRouteImport.update({
     id: '/api/multimodal-tool-result-wire',
@@ -548,6 +554,7 @@ export interface FileRoutesByFullPath {
   '/api/message-ids': typeof ApiMessageIdsRoute
   '/api/middleware-test': typeof ApiMiddlewareTestRoute
   '/api/multimodal-tool-result-wire': typeof ApiMultimodalToolResultWireRoute
+  '/api/non-streaming-run-error': typeof ApiNonStreamingRunErrorRoute
   '/api/openai-completed-response-text': typeof ApiOpenaiCompletedResponseTextRoute
   '/api/openai-shell-skills-wire': typeof ApiOpenaiShellSkillsWireRoute
   '/api/openai-usage-details': typeof ApiOpenaiUsageDetailsRoute
@@ -628,6 +635,7 @@ export interface FileRoutesByTo {
   '/api/message-ids': typeof ApiMessageIdsRoute
   '/api/middleware-test': typeof ApiMiddlewareTestRoute
   '/api/multimodal-tool-result-wire': typeof ApiMultimodalToolResultWireRoute
+  '/api/non-streaming-run-error': typeof ApiNonStreamingRunErrorRoute
   '/api/openai-completed-response-text': typeof ApiOpenaiCompletedResponseTextRoute
   '/api/openai-shell-skills-wire': typeof ApiOpenaiShellSkillsWireRoute
   '/api/openai-usage-details': typeof ApiOpenaiUsageDetailsRoute
@@ -709,6 +717,7 @@ export interface FileRoutesById {
   '/api/message-ids': typeof ApiMessageIdsRoute
   '/api/middleware-test': typeof ApiMiddlewareTestRoute
   '/api/multimodal-tool-result-wire': typeof ApiMultimodalToolResultWireRoute
+  '/api/non-streaming-run-error': typeof ApiNonStreamingRunErrorRoute
   '/api/openai-completed-response-text': typeof ApiOpenaiCompletedResponseTextRoute
   '/api/openai-shell-skills-wire': typeof ApiOpenaiShellSkillsWireRoute
   '/api/openai-usage-details': typeof ApiOpenaiUsageDetailsRoute
@@ -791,6 +800,7 @@ export interface FileRouteTypes {
     | '/api/message-ids'
     | '/api/middleware-test'
     | '/api/multimodal-tool-result-wire'
+    | '/api/non-streaming-run-error'
     | '/api/openai-completed-response-text'
     | '/api/openai-shell-skills-wire'
     | '/api/openai-usage-details'
@@ -871,6 +881,7 @@ export interface FileRouteTypes {
     | '/api/message-ids'
     | '/api/middleware-test'
     | '/api/multimodal-tool-result-wire'
+    | '/api/non-streaming-run-error'
     | '/api/openai-completed-response-text'
     | '/api/openai-shell-skills-wire'
     | '/api/openai-usage-details'
@@ -951,6 +962,7 @@ export interface FileRouteTypes {
     | '/api/message-ids'
     | '/api/middleware-test'
     | '/api/multimodal-tool-result-wire'
+    | '/api/non-streaming-run-error'
     | '/api/openai-completed-response-text'
     | '/api/openai-shell-skills-wire'
     | '/api/openai-usage-details'
@@ -1032,6 +1044,7 @@ export interface RootRouteChildren {
   ApiMessageIdsRoute: typeof ApiMessageIdsRoute
   ApiMiddlewareTestRoute: typeof ApiMiddlewareTestRoute
   ApiMultimodalToolResultWireRoute: typeof ApiMultimodalToolResultWireRoute
+  ApiNonStreamingRunErrorRoute: typeof ApiNonStreamingRunErrorRoute
   ApiOpenaiCompletedResponseTextRoute: typeof ApiOpenaiCompletedResponseTextRoute
   ApiOpenaiShellSkillsWireRoute: typeof ApiOpenaiShellSkillsWireRoute
   ApiOpenaiUsageDetailsRoute: typeof ApiOpenaiUsageDetailsRoute
@@ -1337,6 +1350,13 @@ declare module '@tanstack/react-router' {
       preLoaderRoute: typeof ApiOpenaiCompletedResponseTextRouteImport
       parentRoute: typeof rootRouteImport
     }
+    '/api/non-streaming-run-error': {
+      id: '/api/non-streaming-run-error'
+      path: '/api/non-streaming-run-error'
+      fullPath: '/api/non-streaming-run-error'
+      preLoaderRoute: typeof ApiNonStreamingRunErrorRouteImport
+      parentRoute: typeof rootRouteImport
+    }
     '/api/multimodal-tool-result-wire': {
       id: '/api/multimodal-tool-result-wire'
       path: '/api/multimodal-tool-result-wire'
@@ -1717,6 +1737,7 @@ const rootRouteChildren: RootRouteChildren = {
   ApiMessageIdsRoute: ApiMessageIdsRoute,
   ApiMiddlewareTestRoute: ApiMiddlewareTestRoute,
   ApiMultimodalToolResultWireRoute: ApiMultimodalToolResultWireRoute,
+  ApiNonStreamingRunErrorRoute: ApiNonStreamingRunErrorRoute,
   ApiOpenaiCompletedResponseTextRoute: ApiOpenaiCompletedResponseTextRoute,
   ApiOpenaiShellSkillsWireRoute: ApiOpenaiShellSkillsWireRoute,
   ApiOpenaiUsageDetailsRoute: ApiOpenaiUsageDetailsRoute,

testing/e2e/src/routes/api.non-streaming-run-error.ts

@@ -0,0 +1,54 @@
+import { createFileRoute } from '@tanstack/react-router'
+import { chat } from '@tanstack/ai'
+import { createOpenaiChat } from '@tanstack/ai-openai'
+
+const DUMMY_KEY = 'sk-e2e-test-dummy-key'
+const ERROR_MESSAGE = 'Synthetic upstream failure'
+
+export const Route = createFileRoute('/api/non-streaming-run-error')({
+  server: {
+    handlers: {
+      POST: async () => {
+        const adapter = createOpenaiChat('gpt-5.2', DUMMY_KEY, {
+          fetch: async () =>
+            Response.json(
+              {
+                error: {
+                  message: ERROR_MESSAGE,
+                  type: 'rate_limit_error',
+                  code: 'rate_limit_exceeded',
+                },
+              },
+              { status: 429 },
+            ),
+        })
+
+        try {
+          const text = await chat({
+            adapter,
+            messages: [
+              {
+                role: 'user',
+                content: '[non-streaming-error] trigger the synthetic failure',
+              },
+            ],
+            stream: false,
+          })
+
+          return Response.json({ rejected: false, text })
+        } catch (error) {
+          const message = error instanceof Error ? error.message : String(error)
+          const code =
+            error !== null &&
+            typeof error === 'object' &&
+            'code' in error &&
+            typeof error.code === 'string'
+              ? error.code
+              : null
+
+          return Response.json({ rejected: true, message, code })
+        }
+      },
+    },
+  },
+})

testing/e2e/tests/non-streaming-run-error.spec.ts

@@ -0,0 +1,14 @@
+import { test, expect } from './fixtures'
+
+test('non-streaming chat rejects when the provider stream emits RUN_ERROR', async ({
+  request,
+}) => {
+  const response = await request.post('/api/non-streaming-run-error')
+
+  expect(response.ok()).toBe(true)
+  await expect(response.json()).resolves.toMatchObject({
+    rejected: true,
+    message: expect.stringContaining('Synthetic upstream failure'),
+    code: 'rate_limit_exceeded',
+  })
+})