From 22daae3cf6dd33b0a8adc41fe5996cf29fdbef7f Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Fri, 21 Aug 2026 10:35:49 +0200 Subject: [PATCH] fix: propagate non-streaming run errors # Conflicts: # testing/e2e/src/routeTree.gen.ts --- .changeset/calm-lions-throw.md | 5 ++ .../src/activities/chat/stream/processor.ts | 13 ++--- packages/ai/src/stream-to-response.ts | 5 ++ packages/ai/src/utilities/errors.ts | 23 ++++++++ packages/ai/tests/stream-to-response.test.ts | 31 +++++++++++ testing/e2e/src/routeTree.gen.ts | 21 ++++++++ .../src/routes/api.non-streaming-run-error.ts | 54 +++++++++++++++++++ .../e2e/tests/non-streaming-run-error.spec.ts | 14 +++++ 8 files changed, 156 insertions(+), 10 deletions(-) create mode 100644 .changeset/calm-lions-throw.md create mode 100644 testing/e2e/src/routes/api.non-streaming-run-error.ts create mode 100644 testing/e2e/tests/non-streaming-run-error.spec.ts diff --git a/.changeset/calm-lions-throw.md b/.changeset/calm-lions-throw.md new file mode 100644 index 0000000000..15f1af472f --- /dev/null +++ b/.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`. diff --git a/packages/ai/src/activities/chat/stream/processor.ts b/packages/ai/src/activities/chat/stream/processor.ts index abccc284c5..17ab44883c 100644 --- a/packages/ai/src/activities/chat/stream/processor.ts +++ b/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)) } /** diff --git a/packages/ai/src/stream-to-response.ts b/packages/ai/src/stream-to-response.ts index 6518c4fab8..e49a71fde8 100644 --- a/packages/ai/src/stream-to-response.ts +++ b/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 } diff --git a/packages/ai/src/utilities/errors.ts b/packages/ai/src/utilities/errors.ts index 6254931934..ad6e0f481f 100644 --- a/packages/ai/src/utilities/errors.ts +++ b/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, +): 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 +} diff --git a/packages/ai/tests/stream-to-response.test.ts b/packages/ai/tests/stream-to-response.test.ts index 420f627949..bf53cce0f3 100644 --- a/packages/ai/tests/stream-to-response.test.ts +++ b/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): Promise { 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> = [ diff --git a/testing/e2e/src/routeTree.gen.ts b/testing/e2e/src/routeTree.gen.ts index db1fa5985e..aa27d68dd0 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/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, diff --git a/testing/e2e/src/routes/api.non-streaming-run-error.ts b/testing/e2e/src/routes/api.non-streaming-run-error.ts new file mode 100644 index 0000000000..768ddf6b06 --- /dev/null +++ b/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 }) + } + }, + }, + }, +}) diff --git a/testing/e2e/tests/non-streaming-run-error.spec.ts b/testing/e2e/tests/non-streaming-run-error.spec.ts new file mode 100644 index 0000000000..c718c4782d --- /dev/null +++ b/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', + }) +})