diff --git a/.changeset/recover-streamed-groq-tool-errors.md b/.changeset/recover-streamed-groq-tool-errors.md new file mode 100644 index 0000000000..1932b64355 --- /dev/null +++ b/.changeset/recover-streamed-groq-tool-errors.md @@ -0,0 +1,6 @@ +--- +'@tanstack/openai-base': patch +'@tanstack/ai-groq': patch +--- + +Recover Groq tool calls rejected after streaming begins so agent loops can repair them. diff --git a/packages/ai-groq/tests/groq-adapter.test.ts b/packages/ai-groq/tests/groq-adapter.test.ts index a7b1b6eee4..aec3627d3d 100644 --- a/packages/ai-groq/tests/groq-adapter.test.ts +++ b/packages/ai-groq/tests/groq-adapter.test.ts @@ -558,6 +558,66 @@ describe('Groq AG-UI event emission', () => { } }) + it('emits a non-executable tool error for streamed tool_use_failed', async () => { + const providerError = { + message: 'Failed to call a function. Please adjust your prompt.', + type: 'invalid_request_error', + code: 'tool_use_failed', + failed_generation: JSON.stringify({ + name: 'lookup_weather', + arguments: { location: 'Berlin', units: 'celsius' }, + }), + } + const errorIterable = { + [Symbol.asyncIterator]() { + return { + async next() { + throw Object.assign(new Error(providerError.message), { + code: providerError.code, + error: providerError, + }) + }, + } + }, + } + pendingMockCreate = vi.fn().mockResolvedValue(errorIterable) + + const adapter = createGroqText('llama-3.3-70b-versatile', 'test-api-key') + const chunks: Array = [] + for await (const chunk of adapter.chatStream({ + model: 'llama-3.3-70b-versatile', + messages: [{ role: 'user', content: 'Weather in Berlin?' }], + tools: [weatherTool], + logger: testLogger, + })) { + chunks.push(chunk) + } + + expect(chunks.map((chunk) => chunk.type)).toEqual([ + 'RUN_STARTED', + 'TOOL_CALL_START', + 'TOOL_CALL_ARGS', + 'TOOL_CALL_END', + 'RUN_FINISHED', + ]) + const toolCallEnd = chunks.find((chunk) => chunk.type === 'TOOL_CALL_END') + if (toolCallEnd?.type === 'TOOL_CALL_END') { + expect(toolCallEnd.toolName).toBe('lookup_weather') + expect(toolCallEnd.input).toEqual({ + location: 'Berlin', + units: 'celsius', + }) + expect(toolCallEnd.result).toBe( + JSON.stringify({ error: providerError.message }), + ) + expect(toolCallEnd.state).toBe('output-error') + } + const runFinished = chunks.at(-1) + if (runFinished?.type === 'RUN_FINISHED') { + expect(runFinished.finishReason).toBe('tool_calls') + } + }) + it('emits RUN_ERROR when tool_use_failed has no valid failed generation', async () => { const providerError = { message: 'Failed to call a function. Please adjust your prompt.', diff --git a/packages/openai-base/src/adapters/chat-completions-text.ts b/packages/openai-base/src/adapters/chat-completions-text.ts index fda24530e7..9fea3cd42b 100644 --- a/packages/openai-base/src/adapters/chat-completions-text.ts +++ b/packages/openai-base/src/adapters/chat-completions-text.ts @@ -30,6 +30,13 @@ import type { TextOptions, } from '@tanstack/ai' +type ChatStreamState = { + runId: string + threadId: string + messageId: string + hasEmittedRunStarted: boolean +} + /** * Shared implementation of the OpenAI Chat Completions API. Holds the * stream-accumulator + AG-UI lifecycle logic and calls the OpenAI SDK @@ -94,98 +101,99 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter< yield* this.processStreamChunks(stream, options, aguiState) } catch (error: unknown) { - // Narrow before logging: raw SDK errors can carry request metadata - // (including auth headers) which we must never surface to user loggers. - const errorPayload = toRunErrorPayload( - error, - `${this.name}.chatStream failed`, - ) - const rawEvent = toRunErrorRawEvent(error) + yield* this.handleChatStreamError(error, options, aguiState, 'chatStream') + } + } - // Emit RUN_STARTED if not yet emitted - if (!aguiState.hasEmittedRunStarted) { - aguiState.hasEmittedRunStarted = true - yield { - type: EventType.RUN_STARTED, - runId: aguiState.runId, - threadId: aguiState.threadId, - model: options.model, - timestamp: Date.now(), - parentRunId: options.parentRunId, - } - } + private async *handleChatStreamError( + error: unknown, + options: TextOptions, + aguiState: ChatStreamState, + source: 'chatStream' | 'processStreamChunks', + ): AsyncIterable { + // Narrow before logging: raw SDK errors can carry request metadata + // (including auth headers) which we must never surface to user loggers. + const errorPayload = toRunErrorPayload( + error, + `${this.name}.${source} failed`, + ) + const rawEvent = toRunErrorRawEvent(error) - const rejectedToolCall = this.extractRejectedToolCall( - rawEvent, - errorPayload.message, - ) - if (rejectedToolCall) { - const toolCallId = generateId(this.name) - yield { - type: EventType.TOOL_CALL_START, - toolCallId, - toolCallName: rejectedToolCall.toolName, - toolName: rejectedToolCall.toolName, - parentMessageId: aguiState.messageId, - model: options.model, - timestamp: Date.now(), - } - yield { - type: EventType.TOOL_CALL_ARGS, - toolCallId, - delta: rejectedToolCall.arguments, - args: rejectedToolCall.arguments, - model: options.model, - timestamp: Date.now(), - } - yield { - type: EventType.TOOL_CALL_END, - toolCallId, - toolCallName: rejectedToolCall.toolName, - toolName: rejectedToolCall.toolName, - ...(rejectedToolCall.input !== undefined && { - input: rejectedToolCall.input, - }), - result: JSON.stringify({ error: rejectedToolCall.error }), - state: 'output-error', - model: options.model, - timestamp: Date.now(), - } - yield { - type: EventType.RUN_FINISHED, - runId: aguiState.runId, - threadId: aguiState.threadId, - model: options.model, - timestamp: Date.now(), - finishReason: 'tool_calls', - } - return + if (!aguiState.hasEmittedRunStarted) { + aguiState.hasEmittedRunStarted = true + yield { + type: EventType.RUN_STARTED, + runId: aguiState.runId, + threadId: aguiState.threadId, + model: options.model, + timestamp: Date.now(), + parentRunId: options.parentRunId, } + } - // Emit AG-UI RUN_ERROR. Conditional `code` spread keeps the wire - // shape spec-compliant under `exactOptionalPropertyTypes`: AG-UI's - // `RunErrorEvent.code` is `string?` (absent vs explicit `undefined` - // matter), so we omit the key when there's no code. + const rejectedToolCall = this.extractRejectedToolCall( + rawEvent, + errorPayload.message, + ) + if (rejectedToolCall) { + const toolCallId = generateId(this.name) yield { - type: EventType.RUN_ERROR, + type: EventType.TOOL_CALL_START, + toolCallId, + toolCallName: rejectedToolCall.toolName, + toolName: rejectedToolCall.toolName, + parentMessageId: aguiState.messageId, model: options.model, timestamp: Date.now(), - message: errorPayload.message, - code: errorPayload.code, - // Forward the provider's structured error body so consumers can recover - // the upstream detail the `{ message, code }` payload drops. Omitted - // when the error carried no provider body (see toRunErrorRawEvent). - ...(rawEvent !== undefined && { rawEvent }), - error: { - message: errorPayload.message, - code: errorPayload.code, - }, } + yield { + type: EventType.TOOL_CALL_ARGS, + toolCallId, + delta: rejectedToolCall.arguments, + args: rejectedToolCall.arguments, + model: options.model, + timestamp: Date.now(), + } + yield { + type: EventType.TOOL_CALL_END, + toolCallId, + toolCallName: rejectedToolCall.toolName, + toolName: rejectedToolCall.toolName, + ...(rejectedToolCall.input !== undefined && { + input: rejectedToolCall.input, + }), + result: JSON.stringify({ error: rejectedToolCall.error }), + state: 'output-error', + model: options.model, + timestamp: Date.now(), + } + yield { + type: EventType.RUN_FINISHED, + runId: aguiState.runId, + threadId: aguiState.threadId, + model: options.model, + timestamp: Date.now(), + finishReason: 'tool_calls', + } + return + } - options.logger.errors(`${this.name}.chatStream fatal`, { - error: errorPayload, - source: `${this.name}.chatStream`, - }) + options.logger.errors(`${this.name}.${source} fatal`, { + error: errorPayload, + source: `${this.name}.${source}`, + }) + + yield { + type: EventType.RUN_ERROR, + model: options.model, + timestamp: Date.now(), + message: errorPayload.message, + ...(errorPayload.code !== undefined && { code: errorPayload.code }), + ...(rawEvent !== undefined && { rawEvent }), + error: { + message: errorPayload.message, + ...(errorPayload.code !== undefined && { code: errorPayload.code }), + }, } } @@ -680,12 +688,7 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter< protected async *processStreamChunks( stream: AsyncIterable, options: TextOptions, - aguiState: { - runId: string - threadId: string - messageId: string - hasEmittedRunStarted: boolean - }, + aguiState: ChatStreamState, ): AsyncIterable { let accumulatedContent = '' let hasEmittedTextMessageStart = false @@ -1136,33 +1139,12 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter< } } } catch (error: unknown) { - // Narrow before logging: raw SDK errors can carry request metadata - // (including auth headers) which we must never surface to user loggers. - const errorPayload = toRunErrorPayload( + yield* this.handleChatStreamError( error, - `${this.name}.processStreamChunks failed`, + options, + aguiState, + 'processStreamChunks', ) - const rawEvent = toRunErrorRawEvent(error) - options.logger.errors(`${this.name}.processStreamChunks fatal`, { - error: errorPayload, - source: `${this.name}.processStreamChunks`, - }) - - // Emit AG-UI RUN_ERROR with conditional `code` spread (see chatStream's - // catch block for the rationale). `rawEvent` carries the provider's - // structured error body when present. - yield { - type: EventType.RUN_ERROR, - model: options.model, - timestamp: Date.now(), - message: errorPayload.message, - ...(errorPayload.code !== undefined && { code: errorPayload.code }), - ...(rawEvent !== undefined && { rawEvent }), - error: { - message: errorPayload.message, - ...(errorPayload.code !== undefined && { code: errorPayload.code }), - }, - } } } diff --git a/packages/openai-base/tests/chat-completions-text.test.ts b/packages/openai-base/tests/chat-completions-text.test.ts index a89d8e4e06..a65e35c21a 100644 --- a/packages/openai-base/tests/chat-completions-text.test.ts +++ b/packages/openai-base/tests/chat-completions-text.test.ts @@ -691,20 +691,32 @@ describe('OpenAIBaseChatCompletionsTextAdapter', () => { const adapter = new TestChatCompletionsAdapter(testConfig, 'test-model') const chunks: Array = [] + const errorsSpy = vi.spyOn(testLogger, 'errors') - for await (const chunk of adapter.chatStream({ - logger: testLogger, - model: 'test-model', - messages: [{ role: 'user', content: 'Hello' }], - })) { - chunks.push(chunk) - } + try { + for await (const chunk of adapter.chatStream({ + logger: testLogger, + model: 'test-model', + messages: [{ role: 'user', content: 'Hello' }], + })) { + chunks.push(chunk) + if (chunk.type === EventType.RUN_ERROR) break + } - // Should emit RUN_ERROR - const runErrorChunk = chunks.find((c) => c.type === 'RUN_ERROR') - expect(runErrorChunk).toBeDefined() - if (runErrorChunk?.type === 'RUN_ERROR') { - expect(runErrorChunk.error!.message).toBe('Stream interrupted') + // Should emit RUN_ERROR + const runErrorChunk = chunks.find((c) => c.type === 'RUN_ERROR') + expect(runErrorChunk).toBeDefined() + if (runErrorChunk?.type === 'RUN_ERROR') { + expect(runErrorChunk.error!.message).toBe('Stream interrupted') + } + expect(errorsSpy).toHaveBeenCalledWith( + 'openai-base.processStreamChunks fatal', + expect.objectContaining({ + source: 'openai-base.processStreamChunks', + }), + ) + } finally { + errorsSpy.mockRestore() } })