fix(ai, openai-base, adapters): a watching ChatClient ends each run, a chunk with no choices, and fakeText on request.signal

.changeset/chat-completions-missing-choices.md

@@ -0,0 +1,9 @@
+---
+'@tanstack/openai-base': patch
+'@tanstack/ai-openai': patch
+'@tanstack/ai-groq': patch
+'@tanstack/ai-llmgateway': patch
+'@tanstack/ai-byteplus': patch
+---
+
+Chat Completions adapters skip a stream chunk that has no `choices` field, as some gateways send for the usage trailer. Before, the run failed with `Cannot read properties of undefined (reading '0')`.

.changeset/fake-text-request-signal.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai': patch
+---
+
+`fakeText` from `@tanstack/ai/testing` now stops on `request.signal`, the abort signal `chat()` passes to adapters. Before, it read only `abortController`, so code that reads the adapter stream directly (a harness, for example) saw a paced fake keep streaming after a cancel.

.changeset/watcher-status-ready.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai': patch
+---
+
+A `ChatClient` that only watches a thread (a second tab, with `live: true`) now returns to `ready` after every run. Before, it stayed `streaming` after the second run, because the stream processor ended only the first run it saw.

packages/ai-byteplus/src/adapters/text.ts

@@ -171,7 +171,7 @@ export class BytePlusTextAdapter<
   protected override extractReasoning(
     chunk: OpenAI.Chat.Completions.ChatCompletionChunk,
   ): { text: string } | undefined {
-    const delta = chunk.choices[0]?.delta as
+    const delta = chunk.choices?.[0]?.delta as
       | BytePlusStreamDeltaExtras
       | undefined
     const raw = delta?.reasoning_content
@@ -449,7 +449,7 @@ async function* captureEncryptedContent(
   captured: { encryptedContent?: string },
 ): AsyncIterable<OpenAI.Chat.Completions.ChatCompletionChunk> {
   for await (const chunk of stream) {
-    const delta = chunk.choices[0]?.delta as
+    const delta = chunk.choices?.[0]?.delta as
       | BytePlusStreamDeltaExtras
       | undefined
     const blob = delta?.encrypted_content

packages/ai-client/tests/chat-client.test.ts

@@ -345,6 +345,59 @@ describe('ChatClient', () => {
       expect(adapter.send).toHaveBeenCalled()
     })
 
+    it('returns a client that only watches to ready after each run ends', async () => {
+      // A second tab watches the thread and never sends, so nothing resets
+      // its processor between runs.
+      const threadId = 'thread-watch'
+      const run = (runId: string, answer: string): Array<StreamChunk> => {
+        const base = { threadId, runId, timestamp: 1 }
+        const messageId = `${runId}-answer`
+        return [
+          { ...base, type: EventType.RUN_STARTED },
+          {
+            type: EventType.TEXT_MESSAGE_START,
+            messageId,
+            role: 'assistant',
+            timestamp: 1,
+          },
+          {
+            type: EventType.TEXT_MESSAGE_CONTENT,
+            messageId,
+            delta: answer,
+            timestamp: 1,
+          },
+          { type: EventType.TEXT_MESSAGE_END, messageId, timestamp: 1 },
+          { ...base, type: EventType.RUN_FINISHED, finishReason: 'stop' },
+        ]
+      }
+      let fedAll!: () => void
+      // The feed reaches fedAll() only when the client asks for the chunk
+      // after the last one, so the client has handled every chunk.
+      const handledAll = new Promise<void>((resolve) => (fedAll = resolve))
+      const client = new ChatClient({
+        threadId,
+        connection: {
+          subscribe: (signal) =>
+            (async function* () {
+              yield* run('run-1', 'First answer.')
+              yield* run('run-2', 'Second answer.')
+              fedAll()
+              await new Promise((resolve) => {
+                signal?.addEventListener('abort', resolve, { once: true })
+              })
+            })(),
+          send: () => Promise.resolve(),
+        },
+      })
+
+      client.subscribe()
+      await handledAll
+      expect(JSON.stringify(client.getMessages())).toContain('Second answer.')
+      expect(client.getSessionGenerating()).toBe(false)
+      expect(client.getStatus()).toBe('ready')
+      client.unsubscribe()
+    })
+
     it('should ignore native subscribe/send chunks from a cleared persisted request without runId', async () => {
       let storedMessages: Array<UIMessage> | undefined
       const releaseFirstResponse = createDeferred<void>()

packages/ai-groq/src/adapters/text.ts

@@ -176,7 +176,7 @@ export class GroqTextAdapter<
   protected override extractReasoning(
     chunk: OpenAI.Chat.Completions.ChatCompletionChunk,
   ): { text: string } | undefined {
-    const delta = chunk.choices[0]?.delta as
+    const delta = chunk.choices?.[0]?.delta as
       | { reasoning?: unknown; reasoning_content?: unknown }
       | undefined
     const raw = delta?.reasoning ?? delta?.reasoning_content

packages/ai-llmgateway/src/adapters/text.ts

@@ -106,7 +106,7 @@ export class LLMGatewayTextAdapter<
   protected override extractReasoning(
     chunk: OpenAI.Chat.Completions.ChatCompletionChunk,
   ): { text: string } | undefined {
-    const delta = chunk.choices[0]?.delta as
+    const delta = chunk.choices?.[0]?.delta as
       | { reasoning?: unknown; reasoning_content?: unknown }
       | undefined
     const raw = delta?.reasoning_content ?? delta?.reasoning

packages/ai-openai/src/compatible/adapter.ts

@@ -52,7 +52,7 @@ export class OpenAICompatibleChatAdapter<
   protected override extractReasoning(
     chunk: OpenAI.Chat.Completions.ChatCompletionChunk,
   ): { text: string } | undefined {
-    const delta = chunk.choices[0]?.delta as
+    const delta = chunk.choices?.[0]?.delta as
       | { reasoning?: unknown; reasoning_content?: unknown }
       | undefined
     const raw = delta?.reasoning_content ?? delta?.reasoning

packages/ai-openai/tests/compatible-reasoning.test.ts

@@ -124,4 +124,19 @@ describe('OpenAICompatibleChatAdapter reasoning', () => {
       chunks.some((chunk) => chunk.type === EventType.TEXT_MESSAGE_CONTENT),
     ).toBe(true)
   })
+
+  it('skips a usage chunk that has no choices field', async () => {
+    // Some gateways omit `choices` on the usage trailer.
+    const chunks = await collect([
+      ...textChunks,
+      { usage: { prompt_tokens: 3, completion_tokens: 1, total_tokens: 4 } },
+    ])
+
+    expect(chunks.find((c) => c.type === EventType.RUN_ERROR)).toBeUndefined()
+    expect(chunks.find((c) => c.type === EventType.RUN_FINISHED)).toMatchObject(
+      {
+        usage: { totalTokens: 4 },
+      },
+    )
+  })
 })

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

@@ -2627,6 +2627,9 @@ export class StreamProcessor {
     if (this.activeRuns.size === 0) {
       this.runToolCallIds = []
       this.toolCallsSentToClient.clear()
+      // A new run owes its own stream end. A client that only watches never
+      // resets between runs, so without this only its first run would end.
+      this.streamEndEmitted = false
     }
     this.activeRuns.add(chunk.runId)
   }

packages/ai/src/testing/fake-text.ts

@@ -221,7 +221,8 @@ export class FakeTextAdapter<
   }
 
   async *chatStream(options: TextOptions): AsyncIterable<AdapterYieldChunk> {
-    const signal = options.abortController?.signal
+    // chat() passes the run's signal as `request.signal`.
+    const signal = options.request?.signal ?? options.abortController?.signal
     const runId = options.runId ?? `fake-run-${this.state.callCount + 1}`
     const threadId = options.threadId ?? 'fake-thread'
     const model = this.model

packages/ai/tests/fake-text.test.ts

@@ -242,6 +242,24 @@ describe('fakeText', () => {
     expect(deltas).toEqual(['abcd'])
   })
 
+  it('stops streaming when the request signal chat() passes aborts', async () => {
+    const fake = fakeText({ tokensPerSecond: 100 })
+    fake.setResponses([{ text: 'abcdefghijkl' }])
+    const abortController = new AbortController()
+
+    const deltas: Array<string> = []
+    for await (const event of fake.chatStream(
+      request(hi, { request: { signal: abortController.signal } }),
+    )) {
+      if (event.type === EventType.TEXT_MESSAGE_CONTENT) {
+        deltas.push(event.delta)
+        abortController.abort()
+      }
+    }
+
+    expect(deltas).toEqual(['abcd'])
+  })
+
   it('answers structured output with the next response as JSON', async () => {
     const fake = fakeText()
     fake.setResponses([{ text: '{"city":"Oslo"}' }])

packages/openai-base/src/adapters/chat-completions-text.ts

@@ -505,7 +505,7 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
       )
 
       for await (const chunk of stream) {
-        const choiceForLog = chunk.choices[0]
+        const choiceForLog = chunk.choices?.[0]
         chatOptions.logger.provider(
           `provider=${this.name} finish_reason=${choiceForLog?.finish_reason ?? 'none'} hasContent=${!!choiceForLog?.delta.content} hasUsage=${!!chunk.usage}`,
           { provider: this.name, model: chunk.model },
@@ -574,7 +574,7 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
           }
         }
 
-        const choice = chunk.choices[0]
+        const choice = chunk.choices?.[0]
         if (!choice) continue
         if (choice.finish_reason) finishReason = choice.finish_reason
 
@@ -878,9 +878,8 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
 
     try {
       for await (const chunk of stream) {
-        endsWithUsageOnlyChunk =
-          chunk.choices.length === 0 && chunk.usage != null
-        const choiceForLog = chunk.choices[0]
+        endsWithUsageOnlyChunk = !chunk.choices?.length && chunk.usage != null
+        const choiceForLog = chunk.choices?.[0]
         options.logger.provider(
           `provider=${this.name} finish_reason=${choiceForLog?.finish_reason ?? 'none'} hasContent=${!!choiceForLog?.delta.content} hasToolCalls=${!!choiceForLog?.delta.tool_calls} hasUsage=${!!chunk.usage}`,
           { provider: this.name, model: chunk.model },
@@ -958,7 +957,7 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
           }
         }
 
-        const choice = chunk.choices[0]
+        const choice = chunk.choices?.[0]
 
         if (!choice) continue
 

packages/openai-base/tests/chat-completions-empty-choices.test.ts

@@ -220,4 +220,38 @@ describe('OpenAIBaseChatCompletionsTextAdapter — usage-only terminal chunk', (
       })
     }
   })
+
+  it('skips a usage-only chunk that has no choices field at all', async () => {
+    // Some gateways omit `choices` on the usage trailer instead of sending
+    // `choices: []`.
+    setupMockSdkClient([
+      {
+        id: 'chatcmpl-no-choices',
+        model: 'test-model',
+        choices: [{ delta: { content: 'Hi.' }, finish_reason: 'stop' }],
+      },
+      {
+        id: 'chatcmpl-no-choices',
+        model: 'test-model',
+        usage: { prompt_tokens: 3, completion_tokens: 1, total_tokens: 4 },
+      },
+    ])
+    const adapter = new TestChatCompletionsAdapter(testConfig, 'test-model')
+    const chunks: Array<AdapterYieldChunk> = []
+
+    for await (const chunk of adapter.chatStream({
+      logger: testLogger,
+      model: 'test-model',
+      messages: [{ role: 'user', content: 'Hello' }],
+    })) {
+      chunks.push(chunk)
+    }
+
+    expect(chunks.find((c) => c.type === EventType.RUN_ERROR)).toBeUndefined()
+    const runFinished = chunks.find((c) => c.type === EventType.RUN_FINISHED)
+    expect(runFinished).toMatchObject({
+      finishReason: 'stop',
+      usage: { promptTokens: 3, completionTokens: 1, totalTokens: 4 },
+    })
+  })
 })

packages/openai-base/tests/chat-completions-structured-output-stream.test.ts

@@ -147,6 +147,37 @@ describe('OpenAIBaseChatCompletionsTextAdapter.structuredOutputStream', () => {
       expect(complete!.value.raw).toBe(json)
     })
 
+    it('skips a usage-only chunk that has no choices field at all', async () => {
+      setupStreamingMock([
+        deltaChunk('{"name":"John","age":30}', 'stop'),
+        {
+          id: 'chatcmpl-1',
+          model: 'test-model',
+          usage: { prompt_tokens: 10, completion_tokens: 20, total_tokens: 30 },
+        },
+      ])
+      const adapter = new TestAdapter()
+
+      const chunks = await collect(
+        adapter.structuredOutputStream!({
+          chatOptions: {
+            model: 'test-model',
+            messages: [{ role: 'user', content: 'extract' }],
+            logger: testLogger,
+          },
+          outputSchema: personSchema,
+        }),
+      )
+
+      expect(chunks.find((c) => c.type === 'RUN_ERROR')).toBeUndefined()
+      const complete = chunks.find(
+        (c) =>
+          c.type === 'CUSTOM' &&
+          (c as { name?: string }).name === 'structured-output.complete',
+      ) as { value: { object: unknown } } | undefined
+      expect(complete?.value.object).toEqual({ name: 'John', age: 30 })
+    })
+
     it('timestamps lifecycle events when they are emitted', async () => {
       vi.useFakeTimers()
       try {