fix(ai-groq): recover rejected streamed tool calls

.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.

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<StreamChunk> = []
+    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.',

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<StreamChunk> {
+    // 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<ChatCompletionChunk>,
     options: TextOptions,
-    aguiState: {
-      runId: string
-      threadId: string
-      messageId: string
-      hasEmittedRunStarted: boolean
-    },
+    aguiState: ChatStreamState,
   ): AsyncIterable<StreamChunk> {
     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 }),
-        },
-      }
     }
   }
 

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<StreamChunk> = []
+      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()
       }
     })