diff --git a/.changeset/calm-lions-throw.md b/.changeset/calm-lions-throw.md new file mode 100644 index 000000000..15f1af472 --- /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 2eb6f774b..7cc231fde 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, @@ -1501,15 +1502,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 9850f4d60..e3f9da1c8 100644 --- a/packages/ai/src/stream-to-response.ts +++ b/packages/ai/src/stream-to-response.ts @@ -1,4 +1,5 @@ import { toRunErrorPayload } from './activities/error-payload' +import { runErrorEventToError } from './utilities/errors' import type { StreamChunk } from './types' /** @@ -27,6 +28,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 625493193..ad6e0f481 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 420f62794..bf53cce0f 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 14ffbf330..f5880479d 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/testing/e2e/src/routeTree.gen.ts @@ -33,6 +33,7 @@ import { Route as ApiOpenrouterWebToolsWireRouteImport } from './routes/api.open import { Route as ApiOpenrouterCostRouteImport } from './routes/api.openrouter-cost' import { Route as ApiOpenaiUsageDetailsRouteImport } from './routes/api.openai-usage-details' import { Route as ApiOpenaiShellSkillsWireRouteImport } from './routes/api.openai-shell-skills-wire' +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 ApiMcpTestRouteImport } from './routes/api.mcp-test' @@ -181,6 +182,11 @@ const ApiOpenaiShellSkillsWireRoute = path: '/api/openai-shell-skills-wire', 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', @@ -335,6 +341,7 @@ export interface FileRoutesByFullPath { '/api/mcp-test': typeof ApiMcpTestRoute '/api/middleware-test': typeof ApiMiddlewareTestRoute '/api/multimodal-tool-result-wire': typeof ApiMultimodalToolResultWireRoute + '/api/non-streaming-run-error': typeof ApiNonStreamingRunErrorRoute '/api/openai-shell-skills-wire': typeof ApiOpenaiShellSkillsWireRoute '/api/openai-usage-details': typeof ApiOpenaiUsageDetailsRoute '/api/openrouter-cost': typeof ApiOpenrouterCostRoute @@ -385,6 +392,7 @@ export interface FileRoutesByTo { '/api/mcp-test': typeof ApiMcpTestRoute '/api/middleware-test': typeof ApiMiddlewareTestRoute '/api/multimodal-tool-result-wire': typeof ApiMultimodalToolResultWireRoute + '/api/non-streaming-run-error': typeof ApiNonStreamingRunErrorRoute '/api/openai-shell-skills-wire': typeof ApiOpenaiShellSkillsWireRoute '/api/openai-usage-details': typeof ApiOpenaiUsageDetailsRoute '/api/openrouter-cost': typeof ApiOpenrouterCostRoute @@ -436,6 +444,7 @@ export interface FileRoutesById { '/api/mcp-test': typeof ApiMcpTestRoute '/api/middleware-test': typeof ApiMiddlewareTestRoute '/api/multimodal-tool-result-wire': typeof ApiMultimodalToolResultWireRoute + '/api/non-streaming-run-error': typeof ApiNonStreamingRunErrorRoute '/api/openai-shell-skills-wire': typeof ApiOpenaiShellSkillsWireRoute '/api/openai-usage-details': typeof ApiOpenaiUsageDetailsRoute '/api/openrouter-cost': typeof ApiOpenrouterCostRoute @@ -488,6 +497,7 @@ export interface FileRouteTypes { | '/api/mcp-test' | '/api/middleware-test' | '/api/multimodal-tool-result-wire' + | '/api/non-streaming-run-error' | '/api/openai-shell-skills-wire' | '/api/openai-usage-details' | '/api/openrouter-cost' @@ -538,6 +548,7 @@ export interface FileRouteTypes { | '/api/mcp-test' | '/api/middleware-test' | '/api/multimodal-tool-result-wire' + | '/api/non-streaming-run-error' | '/api/openai-shell-skills-wire' | '/api/openai-usage-details' | '/api/openrouter-cost' @@ -588,6 +599,7 @@ export interface FileRouteTypes { | '/api/mcp-test' | '/api/middleware-test' | '/api/multimodal-tool-result-wire' + | '/api/non-streaming-run-error' | '/api/openai-shell-skills-wire' | '/api/openai-usage-details' | '/api/openrouter-cost' @@ -639,6 +651,7 @@ export interface RootRouteChildren { ApiMcpTestRoute: typeof ApiMcpTestRoute ApiMiddlewareTestRoute: typeof ApiMiddlewareTestRoute ApiMultimodalToolResultWireRoute: typeof ApiMultimodalToolResultWireRoute + ApiNonStreamingRunErrorRoute: typeof ApiNonStreamingRunErrorRoute ApiOpenaiShellSkillsWireRoute: typeof ApiOpenaiShellSkillsWireRoute ApiOpenaiUsageDetailsRoute: typeof ApiOpenaiUsageDetailsRoute ApiOpenrouterCostRoute: typeof ApiOpenrouterCostRoute @@ -824,6 +837,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ApiOpenaiShellSkillsWireRouteImport 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' @@ -1084,6 +1104,7 @@ const rootRouteChildren: RootRouteChildren = { ApiMcpTestRoute: ApiMcpTestRoute, ApiMiddlewareTestRoute: ApiMiddlewareTestRoute, ApiMultimodalToolResultWireRoute: ApiMultimodalToolResultWireRoute, + ApiNonStreamingRunErrorRoute: ApiNonStreamingRunErrorRoute, ApiOpenaiShellSkillsWireRoute: ApiOpenaiShellSkillsWireRoute, ApiOpenaiUsageDetailsRoute: ApiOpenaiUsageDetailsRoute, ApiOpenrouterCostRoute: ApiOpenrouterCostRoute, 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 000000000..768ddf6b0 --- /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 000000000..c718c4782 --- /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', + }) +})