From 5f41204a442e189f6039a84ad7b9346942b0cf4a Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 12 Oct 2026 19:19:48 +0200 Subject: [PATCH 1/3] fix(ai): return a watching ChatClient to ready after every run A client that only subscribes never resets its stream processor between runs, so onStreamEnd fired for the first run only and the client stayed 'streaming' after the second. A run that starts with no other run active now owes its own stream end. --- .changeset/watcher-status-ready.md | 5 + packages/ai-client/tests/chat-client.test.ts | 53 +++++++++ .../src/activities/chat/stream/processor.ts | 3 + testing/e2e/src/routeTree.gen.ts | 21 ++++ .../src/routes/chat-client-watcher-status.tsx | 104 ++++++++++++++++++ .../tests/chat-client-watcher-status.spec.ts | 17 +++ 6 files changed, 203 insertions(+) create mode 100644 .changeset/watcher-status-ready.md create mode 100644 testing/e2e/src/routes/chat-client-watcher-status.tsx create mode 100644 testing/e2e/tests/chat-client-watcher-status.spec.ts diff --git a/.changeset/watcher-status-ready.md b/.changeset/watcher-status-ready.md new file mode 100644 index 0000000000..5ca19234dc --- /dev/null +++ b/.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. diff --git a/packages/ai-client/tests/chat-client.test.ts b/packages/ai-client/tests/chat-client.test.ts index 7340cb015f..84e4f09bfa 100644 --- a/packages/ai-client/tests/chat-client.test.ts +++ b/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 => { + 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((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 | undefined const releaseFirstResponse = createDeferred() diff --git a/packages/ai/src/activities/chat/stream/processor.ts b/packages/ai/src/activities/chat/stream/processor.ts index 05faf7b5a9..ac7f97dd6f 100644 --- a/packages/ai/src/activities/chat/stream/processor.ts +++ b/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) } diff --git a/testing/e2e/src/routeTree.gen.ts b/testing/e2e/src/routeTree.gen.ts index 11dec53912..87143784e5 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/testing/e2e/src/routeTree.gen.ts @@ -14,6 +14,7 @@ import { Route as ActivityTestRouteImport } from './routes/activity-test' import { Route as ByokRouteImport } from './routes/byok' import { Route as ChatClientDefaultBridgeRouteImport } from './routes/chat-client-default-bridge' import { Route as ChatClientStreamProcessingRouteImport } from './routes/chat-client-stream-processing' +import { Route as ChatClientWatcherStatusRouteImport } from './routes/chat-client-watcher-status' import { Route as ClientMountHydrateRouteImport } from './routes/client-mount-hydrate' import { Route as DevtoolsChatRouteImport } from './routes/devtools-chat' import { Route as DevtoolsGenerationHooksRouteImport } from './routes/devtools-generation-hooks' @@ -193,6 +194,11 @@ const ChatClientStreamProcessingRoute = path: '/chat-client-stream-processing', getParentRoute: () => rootRouteImport, } as any) +const ChatClientWatcherStatusRoute = ChatClientWatcherStatusRouteImport.update({ + id: '/chat-client-watcher-status', + path: '/chat-client-watcher-status', + getParentRoute: () => rootRouteImport, +} as any) const ClientMountHydrateRoute = ClientMountHydrateRouteImport.update({ id: '/client-mount-hydrate', path: '/client-mount-hydrate', @@ -996,6 +1002,7 @@ export interface FileRoutesByFullPath { '/byok': typeof ByokRoute '/chat-client-default-bridge': typeof ChatClientDefaultBridgeRoute '/chat-client-stream-processing': typeof ChatClientStreamProcessingRoute + '/chat-client-watcher-status': typeof ChatClientWatcherStatusRoute '/client-mount-hydrate': typeof ClientMountHydrateRoute '/devtools-chat': typeof DevtoolsChatRoute '/devtools-generation-hooks': typeof DevtoolsGenerationHooksRoute @@ -1155,6 +1162,7 @@ export interface FileRoutesByTo { '/byok': typeof ByokRoute '/chat-client-default-bridge': typeof ChatClientDefaultBridgeRoute '/chat-client-stream-processing': typeof ChatClientStreamProcessingRoute + '/chat-client-watcher-status': typeof ChatClientWatcherStatusRoute '/client-mount-hydrate': typeof ClientMountHydrateRoute '/devtools-chat': typeof DevtoolsChatRoute '/devtools-generation-hooks': typeof DevtoolsGenerationHooksRoute @@ -1315,6 +1323,7 @@ export interface FileRoutesById { '/byok': typeof ByokRoute '/chat-client-default-bridge': typeof ChatClientDefaultBridgeRoute '/chat-client-stream-processing': typeof ChatClientStreamProcessingRoute + '/chat-client-watcher-status': typeof ChatClientWatcherStatusRoute '/client-mount-hydrate': typeof ClientMountHydrateRoute '/devtools-chat': typeof DevtoolsChatRoute '/devtools-generation-hooks': typeof DevtoolsGenerationHooksRoute @@ -1476,6 +1485,7 @@ export interface FileRouteTypes { | '/byok' | '/chat-client-default-bridge' | '/chat-client-stream-processing' + | '/chat-client-watcher-status' | '/client-mount-hydrate' | '/devtools-chat' | '/devtools-generation-hooks' @@ -1635,6 +1645,7 @@ export interface FileRouteTypes { | '/byok' | '/chat-client-default-bridge' | '/chat-client-stream-processing' + | '/chat-client-watcher-status' | '/client-mount-hydrate' | '/devtools-chat' | '/devtools-generation-hooks' @@ -1794,6 +1805,7 @@ export interface FileRouteTypes { | '/byok' | '/chat-client-default-bridge' | '/chat-client-stream-processing' + | '/chat-client-watcher-status' | '/client-mount-hydrate' | '/devtools-chat' | '/devtools-generation-hooks' @@ -1954,6 +1966,7 @@ export interface RootRouteChildren { ByokRoute: typeof ByokRoute ChatClientDefaultBridgeRoute: typeof ChatClientDefaultBridgeRoute ChatClientStreamProcessingRoute: typeof ChatClientStreamProcessingRoute + ChatClientWatcherStatusRoute: typeof ChatClientWatcherStatusRoute ClientMountHydrateRoute: typeof ClientMountHydrateRoute DevtoolsChatRoute: typeof DevtoolsChatRoute DevtoolsGenerationHooksRoute: typeof DevtoolsGenerationHooksRoute @@ -2140,6 +2153,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ChatClientStreamProcessingRouteImport parentRoute: typeof rootRouteImport } + '/chat-client-watcher-status': { + id: '/chat-client-watcher-status' + path: '/chat-client-watcher-status' + fullPath: '/chat-client-watcher-status' + preLoaderRoute: typeof ChatClientWatcherStatusRouteImport + parentRoute: typeof rootRouteImport + } '/client-mount-hydrate': { id: '/client-mount-hydrate' path: '/client-mount-hydrate' @@ -3271,6 +3291,7 @@ const rootRouteChildren: RootRouteChildren = { ByokRoute: ByokRoute, ChatClientDefaultBridgeRoute: ChatClientDefaultBridgeRoute, ChatClientStreamProcessingRoute: ChatClientStreamProcessingRoute, + ChatClientWatcherStatusRoute: ChatClientWatcherStatusRoute, ClientMountHydrateRoute: ClientMountHydrateRoute, DevtoolsChatRoute: DevtoolsChatRoute, DevtoolsGenerationHooksRoute: DevtoolsGenerationHooksRoute, diff --git a/testing/e2e/src/routes/chat-client-watcher-status.tsx b/testing/e2e/src/routes/chat-client-watcher-status.tsx new file mode 100644 index 0000000000..2e46f1d217 --- /dev/null +++ b/testing/e2e/src/routes/chat-client-watcher-status.tsx @@ -0,0 +1,104 @@ +import { useEffect, useMemo, useState } from 'react' +import { createFileRoute } from '@tanstack/react-router' +import { EventType } from '@tanstack/ai' +import { useChat } from '@tanstack/ai-react' +import type { StreamChunk } from '@tanstack/ai' +import type { SubscribeConnectionAdapter } from '@tanstack/ai-react' + +export const Route = createFileRoute('/chat-client-watcher-status')({ + component: ChatClientWatcherStatusPage, +}) + +const threadId = 'watched-thread' + +function runChunks(runId: string, answer: string): Array { + const base = { threadId, runId, timestamp: Date.now() } + const messageId = `${runId}-answer` + return [ + { ...base, type: EventType.RUN_STARTED }, + { + type: EventType.TEXT_MESSAGE_START, + messageId, + role: 'assistant', + timestamp: base.timestamp, + }, + { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId, + delta: answer, + timestamp: base.timestamp, + }, + { type: EventType.TEXT_MESSAGE_END, messageId, timestamp: base.timestamp }, + { ...base, type: EventType.RUN_FINISHED, finishReason: 'stop' }, + ] +} + +// Another tab sends. This page only watches the thread's events. +function createWatchedConnection() { + const queue: Array = [] + let wake: (() => void) | undefined + const connection: SubscribeConnectionAdapter = { + subscribe(signal) { + return (async function* () { + while (!signal?.aborted) { + if (queue.length === 0) { + await new Promise((resolve) => { + wake = resolve + signal?.addEventListener('abort', () => resolve(), { + once: true, + }) + }) + wake = undefined + } + let chunk = queue.shift() + while (chunk !== undefined) { + yield chunk + chunk = queue.shift() + } + } + })() + }, + send: () => Promise.resolve(), + } + const push = (chunks: Array) => { + queue.push(...chunks) + wake?.() + } + return { connection, push } +} + +function ChatClientWatcherStatusPage() { + const watched = useMemo(createWatchedConnection, []) + const [hydrated, setHydrated] = useState(false) + const [runs, setRuns] = useState(0) + const { messages, status } = useChat({ + connection: watched.connection, + threadId, + live: true, + }) + + useEffect(() => { + setHydrated(true) + }, []) + + const answers = messages.filter((message) => message.role === 'assistant') + + return ( +
+ + {status} + {answers.length} +
+ ) +} diff --git a/testing/e2e/tests/chat-client-watcher-status.spec.ts b/testing/e2e/tests/chat-client-watcher-status.spec.ts new file mode 100644 index 0000000000..3bb7ec257b --- /dev/null +++ b/testing/e2e/tests/chat-client-watcher-status.spec.ts @@ -0,0 +1,17 @@ +import { expect, test } from '@playwright/test' + +test('a ChatClient that only watches returns to ready after every run', async ({ + page, +}) => { + await page.goto('/chat-client-watcher-status') + const run = page.getByTestId('run') + await expect(run).toBeEnabled() + + await run.click() + await expect(page.getByTestId('answers')).toHaveText('1') + await expect(page.getByTestId('status')).toHaveText('ready') + + await run.click() + await expect(page.getByTestId('answers')).toHaveText('2') + await expect(page.getByTestId('status')).toHaveText('ready') +}) From cbbe8bb6f567bc265e3175296b7f87e920a10c13 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 12 Oct 2026 19:19:48 +0200 Subject: [PATCH 2/3] fix(openai-base): skip a Chat Completions chunk with no choices field Some gateways send the usage trailer without 'choices'. The log line read chunk.choices[0] before the adapter's own guard, so the run failed with a TypeError. Both stream paths now read choices optionally. --- .../chat-completions-missing-choices.md | 9 +++++ packages/ai-byteplus/src/adapters/text.ts | 4 +-- packages/ai-groq/src/adapters/text.ts | 2 +- packages/ai-llmgateway/src/adapters/text.ts | 2 +- packages/ai-openai/src/compatible/adapter.ts | 2 +- .../tests/compatible-reasoning.test.ts | 15 ++++++++ .../src/adapters/chat-completions-text.ts | 11 +++--- .../chat-completions-empty-choices.test.ts | 34 +++++++++++++++++++ ...mpletions-structured-output-stream.test.ts | 31 +++++++++++++++++ .../api.chat-completions-incomplete-stream.ts | 4 ++- ...chat-completions-incomplete-stream.spec.ts | 18 ++++++++++ 11 files changed, 120 insertions(+), 12 deletions(-) create mode 100644 .changeset/chat-completions-missing-choices.md diff --git a/.changeset/chat-completions-missing-choices.md b/.changeset/chat-completions-missing-choices.md new file mode 100644 index 0000000000..6e671aac4b --- /dev/null +++ b/.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')`. diff --git a/packages/ai-byteplus/src/adapters/text.ts b/packages/ai-byteplus/src/adapters/text.ts index e5c937748b..13a4b930a8 100644 --- a/packages/ai-byteplus/src/adapters/text.ts +++ b/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 { 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 diff --git a/packages/ai-groq/src/adapters/text.ts b/packages/ai-groq/src/adapters/text.ts index 223c1149c2..0b7f778a52 100644 --- a/packages/ai-groq/src/adapters/text.ts +++ b/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 diff --git a/packages/ai-llmgateway/src/adapters/text.ts b/packages/ai-llmgateway/src/adapters/text.ts index 37729ebd31..a6e03e5d9f 100644 --- a/packages/ai-llmgateway/src/adapters/text.ts +++ b/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 diff --git a/packages/ai-openai/src/compatible/adapter.ts b/packages/ai-openai/src/compatible/adapter.ts index dd608dbf06..5deb44e087 100644 --- a/packages/ai-openai/src/compatible/adapter.ts +++ b/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 diff --git a/packages/ai-openai/tests/compatible-reasoning.test.ts b/packages/ai-openai/tests/compatible-reasoning.test.ts index faf2227985..92a4d6b75c 100644 --- a/packages/ai-openai/tests/compatible-reasoning.test.ts +++ b/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 }, + }, + ) + }) }) diff --git a/packages/openai-base/src/adapters/chat-completions-text.ts b/packages/openai-base/src/adapters/chat-completions-text.ts index 69d8af6d59..6c720d27a6 100644 --- a/packages/openai-base/src/adapters/chat-completions-text.ts +++ b/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 diff --git a/packages/openai-base/tests/chat-completions-empty-choices.test.ts b/packages/openai-base/tests/chat-completions-empty-choices.test.ts index 4ab0479693..dd6209c497 100644 --- a/packages/openai-base/tests/chat-completions-empty-choices.test.ts +++ b/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 = [] + + 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 }, + }) + }) }) diff --git a/packages/openai-base/tests/chat-completions-structured-output-stream.test.ts b/packages/openai-base/tests/chat-completions-structured-output-stream.test.ts index 59f4933b45..be2ee84657 100644 --- a/packages/openai-base/tests/chat-completions-structured-output-stream.test.ts +++ b/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 { diff --git a/testing/e2e/src/routes/api.chat-completions-incomplete-stream.ts b/testing/e2e/src/routes/api.chat-completions-incomplete-stream.ts index bdc6bdf1b9..ed8ff9680b 100644 --- a/testing/e2e/src/routes/api.chat-completions-incomplete-stream.ts +++ b/testing/e2e/src/routes/api.chat-completions-incomplete-stream.ts @@ -66,7 +66,8 @@ export const Route = createFileRoute('/api/chat-completions-incomplete-stream')( }, { ...envelope, - choices: [], + // Some gateways omit `choices` on the usage trailer. + ...(scenario === 'no-choices' ? {} : { choices: [] }), usage: { prompt_tokens: 10, completion_tokens: 2, @@ -84,6 +85,7 @@ export const Route = createFileRoute('/api/chat-completions-incomplete-stream')( // Finish a follow-up request if a regression executes the tool. const complete = scenario === 'complete' || + scenario === 'no-choices' || scenario === 'object-content' || scenario === 'unknown-finish' || requests++ > 0 diff --git a/testing/e2e/tests/chat-completions-incomplete-stream.spec.ts b/testing/e2e/tests/chat-completions-incomplete-stream.spec.ts index 5405b865dc..8bd71b3326 100644 --- a/testing/e2e/tests/chat-completions-incomplete-stream.spec.ts +++ b/testing/e2e/tests/chat-completions-incomplete-stream.spec.ts @@ -61,6 +61,24 @@ test('Chat Completions keeps usage after a normal finish reason', async ({ expect(result.events).not.toContain('RUN_ERROR') }) +test('Chat Completions skips a usage chunk that has no choices field', async ({ + request, +}) => { + const response = await request.post( + '/api/chat-completions-incomplete-stream?scenario=no-choices', + ) + expect(response.ok()).toBe(true) + const result = await response.json() + expect(result).toMatchObject({ + text: 'partial', + finishCount: 1, + errorCount: 0, + totalTokens: 12, + }) + expect(result.events.at(-1)).toBe('RUN_FINISHED') + expect(result.events).not.toContain('RUN_ERROR') +}) + test('Chat Completions rejects object text content instead of streaming it', async ({ request, }) => { From 012f93eab19e9e07466e487e5319c58ea35bdd55 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 12 Oct 2026 19:19:48 +0200 Subject: [PATCH 3/3] fix(ai): stop fakeText streaming on the signal chat() passes chat() passes the run's signal as request.signal, but fakeText read only abortController, so a paced fake kept streaming after a cancel. --- .changeset/fake-text-request-signal.md | 5 +++++ packages/ai/src/testing/fake-text.ts | 3 ++- packages/ai/tests/fake-text.test.ts | 18 ++++++++++++++++++ 3 files changed, 25 insertions(+), 1 deletion(-) create mode 100644 .changeset/fake-text-request-signal.md diff --git a/.changeset/fake-text-request-signal.md b/.changeset/fake-text-request-signal.md new file mode 100644 index 0000000000..1883b898f0 --- /dev/null +++ b/.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. diff --git a/packages/ai/src/testing/fake-text.ts b/packages/ai/src/testing/fake-text.ts index b54e4d6ed5..49ac19b41e 100644 --- a/packages/ai/src/testing/fake-text.ts +++ b/packages/ai/src/testing/fake-text.ts @@ -221,7 +221,8 @@ export class FakeTextAdapter< } async *chatStream(options: TextOptions): AsyncIterable { - 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 diff --git a/packages/ai/tests/fake-text.test.ts b/packages/ai/tests/fake-text.test.ts index 8898e6f994..89f03f1057 100644 --- a/packages/ai/tests/fake-text.test.ts +++ b/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 = [] + 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"}' }])