From eb4dcd14b5454398bb095dcf3ed708d4f6ec0cbd Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 14:51:15 +0200 Subject: [PATCH 01/23] feat(ai-client): add message-queue config types --- packages/ai-client/src/types.ts | 69 +++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/packages/ai-client/src/types.ts b/packages/ai-client/src/types.ts index 25e07c42b..a7a6d0a47 100644 --- a/packages/ai-client/src/types.ts +++ b/packages/ai-client/src/types.ts @@ -144,6 +144,62 @@ export interface MultimodalContent { id?: string } +/** + * Action taken when `sendMessage` is called while a stream is in flight. + * - `queue`: hold the message; it auto-sends when the stream settles. + * - `drop`: ignore the send. + * - `interrupt`: abort the current stream (like `stop()`) and send now. + */ +export type WhenBusy = 'queue' | 'drop' | 'interrupt' + +/** + * A user message held in the send queue while a stream is active. + * Rendered separately from `messages`; cancellable via `cancelQueued(id)` + * until it drains. + */ +export interface QueuedMessage { + id: string + content: string | MultimodalContent + createdAt: number +} + +/** + * Declarative queue policy. + */ +export interface QueueConfig { + /** Action when a stream is in flight. Default `'queue'`. */ + whenBusy?: WhenBusy + /** + * How queued items leave the queue. + * - `'fifo'`: one at a time, in order (default). + * - `'batch'`: merge all queued items into one send when the stream settles. + */ + drain?: 'fifo' | 'batch' + /** Max queued items. Unlimited when omitted. */ + maxSize?: number + /** Behavior when `maxSize` is reached. Default `'reject'`. */ + onOverflow?: 'reject' | 'drop-oldest' +} + +/** + * Escape hatch: decide the action for a single send. Drain stays FIFO for the + * function form (no `batch` via function). + */ +export type QueueStrategy = (ctx: { + pending: QueuedMessage + isStreaming: boolean + queued: ReadonlyArray +}) => { action: 'send' | 'enqueue' | 'drop' | 'interrupt' } + +/** A `WhenBusy` shorthand, a full config, or a strategy function. */ +export type QueueOption = WhenBusy | QueueConfig | QueueStrategy + +/** Per-call overrides for `sendMessage`. */ +export interface SendMessageOptions { + /** Overrides the configured `whenBusy` for this one send. */ + whenBusy?: WhenBusy +} + /** * Message parts - building blocks of UIMessage */ @@ -480,6 +536,19 @@ export interface ChatClientBaseOptions< */ onSessionGeneratingChange?: (isGenerating: boolean) => void + /** + * Policy for messages sent while a stream is in flight. Accepts a + * `WhenBusy` string, a `QueueConfig`, or a `QueueStrategy` function. + * Default: `{ whenBusy: 'queue', drain: 'fifo' }`. + */ + queue?: QueueOption + + /** + * Callback when the pending send queue changes (enqueue, cancel, drain, + * or flush). + */ + onQueueChange?: (queue: Array) => void + /** * Callback when a custom event is received from a server-side tool. * Custom events are emitted by tools using `context.emitCustomEvent()` during execution. From 7da583f0bb76c9ae9af9274648fdd3f6bf54dae9 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 14:53:06 +0200 Subject: [PATCH 02/23] feat(ai-client): export message-queue public types --- packages/ai-client/src/index.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/ai-client/src/index.ts b/packages/ai-client/src/index.ts index bd9270ff6..62f7d28b9 100644 --- a/packages/ai-client/src/index.ts +++ b/packages/ai-client/src/index.ts @@ -35,6 +35,12 @@ export type { ChatTransport, DistributedOmit, MultimodalContent, + QueuedMessage, + WhenBusy, + QueueConfig, + QueueStrategy, + QueueOption, + SendMessageOptions, } from './types' // Generation client types export type { From e29415978f02ec955ae10fc26e1c09dab7b72699 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 14:53:49 +0200 Subject: [PATCH 03/23] feat(ai-client): add message-queue config normalization + tests --- packages/ai-client/src/chat-client.ts | 32 +++++++++++++++++ .../ai-client/tests/chat-client-queue.test.ts | 35 +++++++++++++++++++ 2 files changed, 67 insertions(+) create mode 100644 packages/ai-client/tests/chat-client-queue.test.ts diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index d5ed91a8b..40821592e 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -39,8 +39,13 @@ import type { ConnectionStatus, MessagePart, MultimodalContent, + QueueOption, + QueuedMessage, + QueueStrategy, + SendMessageOptions, ToolCallPart, UIMessage, + WhenBusy, } from './types' type ChatClientUpdateOptionsWithoutContext< @@ -89,6 +94,33 @@ function resolveTransport(transport: { throw new Error('ChatClient: either `connection` or `fetcher` is required.') } +export interface NormalizedQueueConfig { + whenBusy: WhenBusy + drain: 'fifo' | 'batch' + onOverflow: 'reject' | 'drop-oldest' + maxSize?: number + strategy?: QueueStrategy +} + +export function normalizeQueueOption( + option: QueueOption | undefined, +): NormalizedQueueConfig { + const base: NormalizedQueueConfig = { + whenBusy: 'queue', + drain: 'fifo', + onOverflow: 'reject', + } + if (!option) return base + if (typeof option === 'string') return { ...base, whenBusy: option } + if (typeof option === 'function') return { ...base, strategy: option } + return { + whenBusy: option.whenBusy ?? 'queue', + drain: option.drain ?? 'fifo', + onOverflow: option.onOverflow ?? 'reject', + ...(option.maxSize !== undefined ? { maxSize: option.maxSize } : {}), + } +} + export class ChatClient< TTools extends ReadonlyArray = any, TContext = unknown, diff --git a/packages/ai-client/tests/chat-client-queue.test.ts b/packages/ai-client/tests/chat-client-queue.test.ts new file mode 100644 index 000000000..49f04989a --- /dev/null +++ b/packages/ai-client/tests/chat-client-queue.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from 'vitest' +import { normalizeQueueOption } from '../src/chat-client' + +describe('normalizeQueueOption', () => { + it('defaults to queue + fifo + reject', () => { + expect(normalizeQueueOption(undefined)).toEqual({ + whenBusy: 'queue', + drain: 'fifo', + onOverflow: 'reject', + }) + }) + + it('treats a string as whenBusy shorthand', () => { + expect(normalizeQueueOption('interrupt')).toMatchObject({ + whenBusy: 'interrupt', + drain: 'fifo', + }) + }) + + it('carries a function as strategy and forces fifo', () => { + const fn = () => ({ action: 'enqueue' as const }) + const cfg = normalizeQueueOption(fn) + expect(cfg.strategy).toBe(fn) + expect(cfg.drain).toBe('fifo') + }) + + it('merges a config object over defaults', () => { + expect(normalizeQueueOption({ whenBusy: 'drop', maxSize: 3 })).toEqual({ + whenBusy: 'drop', + drain: 'fifo', + onOverflow: 'reject', + maxSize: 3, + }) + }) +}) From 9bd6bcbec54adf2b0f6975fd7e558e7c11d273c0 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:02:23 +0200 Subject: [PATCH 04/23] feat(ai-client): enqueue/cancel/flush + whenBusy policy in ChatClient --- packages/ai-client/src/chat-client.ts | 112 ++++++++++++++++-- .../ai-client/tests/chat-client-queue.test.ts | 99 +++++++++++++++- 2 files changed, 201 insertions(+), 10 deletions(-) diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index 40821592e..c198cfb66 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -40,8 +40,8 @@ import type { MessagePart, MultimodalContent, QueueOption, - QueuedMessage, QueueStrategy, + QueuedMessage, SendMessageOptions, ToolCallPart, UIMessage, @@ -142,6 +142,9 @@ export class ChatClient< private forwardedPropsOption: Record = {} private context: TContext | undefined = undefined private pendingMessageBody: Record | undefined = undefined + private readonly queueConfig: NormalizedQueueConfig + private messageQueue: Array }> = + [] private isLoading = false private isSubscribed = false private error: Error | undefined = undefined @@ -191,6 +194,7 @@ export class ChatClient< onSubscriptionChange: (isSubscribed: boolean) => void onConnectionStatusChange: (status: ConnectionStatus) => void onSessionGeneratingChange: (isGenerating: boolean) => void + onQueueChange: (queue: Array) => void onCustomEvent: ( eventType: string, data: unknown, @@ -217,6 +221,7 @@ export class ChatClient< this.bodyOption = options.body || {} this.forwardedPropsOption = options.forwardedProps || {} this.context = options.context + this.queueConfig = normalizeQueueOption(options.queue) this.connection = normalizeConnectionAdapter(resolveTransport(options)) // Build client tools map @@ -247,6 +252,7 @@ export class ChatClient< options.onConnectionStatusChange || (() => {}), onSessionGeneratingChange: options.onSessionGeneratingChange || (() => {}), + onQueueChange: options.onQueueChange || (() => {}), onCustomEvent: options.onCustomEvent || (() => {}), }, } @@ -803,28 +809,86 @@ export class ChatClient< async sendMessage( content: string | MultimodalContent, body?: Record, + sendOptions?: SendMessageOptions, ): Promise { this.mountDevtools() const emptyMessage = typeof content === 'string' && !content.trim() - if (emptyMessage || this.isLoading) { + if (emptyMessage) { return } - // Normalize input to extract content, id, and validate - const normalizedContent = this.normalizeMessageInput(content) - // Store the per-message body for use in streamResponse - this.pendingMessageBody = body + if (this.isLoading) { + const action = this.decideWhenBusy(content, sendOptions) + if (action === 'drop') { + return + } + if (action === 'enqueue') { + this.enqueueMessage(content, body) + return + } + // 'interrupt': abort the current stream, then fall through to send now. + this.cancelInFlightStream({ setReadyStatus: true }) + this.resetSessionGenerating() + } - // Add user message via processor + const normalizedContent = this.normalizeMessageInput(content) + this.pendingMessageBody = body const userMessage = this.processor.addUserMessage( normalizedContent.content, normalizedContent.id, ) this.events.messageSent(userMessage.id, normalizedContent.content) - await this.streamResponse() } + /** + * Resolve the effective action for a send that arrives while streaming. + * A strategy that returns `'send'` while streaming is coerced to + * `'enqueue'` — you cannot start a second concurrent stream. + */ + private decideWhenBusy( + content: string | MultimodalContent, + sendOptions?: SendMessageOptions, + ): 'drop' | 'enqueue' | 'interrupt' { + if (sendOptions?.whenBusy) { + return sendOptions.whenBusy === 'queue' ? 'enqueue' : sendOptions.whenBusy + } + const { strategy, whenBusy } = this.queueConfig + if (strategy) { + const { action } = strategy({ + pending: { + id: this.generateUniqueId('queued'), + content, + createdAt: Date.now(), + }, + isStreaming: true, + queued: this.getQueue(), + }) + return action === 'send' ? 'enqueue' : action + } + return whenBusy === 'queue' ? 'enqueue' : whenBusy + } + + private enqueueMessage( + content: string | MultimodalContent, + body?: Record, + ): void { + const { maxSize, onOverflow } = this.queueConfig + if (maxSize !== undefined && this.messageQueue.length >= maxSize) { + if (onOverflow === 'reject') { + return + } + this.messageQueue.shift() // drop-oldest + } + this.messageQueue.push({ + id: this.generateUniqueId('queued'), + content, + createdAt: Date.now(), + ...(body !== undefined ? { body } : {}), + }) + this.emitQueueChange() + } + /** * Normalize the message input to extract content and optional id. * Trims string content automatically. @@ -1408,6 +1472,38 @@ export class ChatClient< return this.processor.getMessages() as Array> } + /** + * Get the current send queue (messages held while a stream was in flight). + */ + getQueue(): Array { + return this.messageQueue.map(({ id, content, createdAt }) => ({ + id, + content, + createdAt, + })) + } + + private emitQueueChange(): void { + this.callbacksRef.current.onQueueChange(this.getQueue()) + this.devtoolsBridge.emitSnapshot() + } + + /** + * Remove a queued message by id before it drains. + */ + cancelQueued(id: string): void { + const index = this.messageQueue.findIndex((m) => m.id === id) + if (index === -1) return + this.messageQueue.splice(index, 1) + this.emitQueueChange() + } + + protected flushQueue(): void { + if (this.messageQueue.length === 0) return + this.messageQueue = [] + this.emitQueueChange() + } + /** * Get loading state */ diff --git a/packages/ai-client/tests/chat-client-queue.test.ts b/packages/ai-client/tests/chat-client-queue.test.ts index 49f04989a..52f46467c 100644 --- a/packages/ai-client/tests/chat-client-queue.test.ts +++ b/packages/ai-client/tests/chat-client-queue.test.ts @@ -1,5 +1,7 @@ -import { describe, expect, it } from 'vitest' -import { normalizeQueueOption } from '../src/chat-client' +import { describe, expect, it, vi } from 'vitest' +import { ChatClient, normalizeQueueOption } from '../src/chat-client' +import { createTextChunks } from './test-utils' +import type { ConnectConnectionAdapter } from '../src/connection-adapters' describe('normalizeQueueOption', () => { it('defaults to queue + fifo + reject', () => { @@ -33,3 +35,96 @@ describe('normalizeQueueOption', () => { }) }) }) + +/** + * Creates a deferred promise. `resolve` releases anything awaiting `promise`. + */ +function createDeferred() { + let resolve!: (value: T) => void + const promise = new Promise((promiseResolve) => { + resolve = promiseResolve + }) + return { promise, resolve } +} + +/** + * A `ConnectConnectionAdapter` whose stream stays open (keeping `isLoading` + * true) until `release()` is called, at which point it finishes with a + * simple text response. + */ +function createHoldingConnection(): { + connection: ConnectConnectionAdapter + release: () => void +} { + const deferred = createDeferred() + const connection: ConnectConnectionAdapter = { + async *connect() { + await deferred.promise + yield* createTextChunks('done', 'msg-1') + }, + } + return { connection, release: () => deferred.resolve() } +} + +describe('ChatClient message queue', () => { + it('enqueues (not drops) a send while streaming and reports via onQueueChange', async () => { + const { connection, release } = createHoldingConnection() + const onQueueChange = vi.fn() + const client = new ChatClient({ connection, onQueueChange }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + await client.sendMessage('second') + + const queue = client.getQueue() + expect(queue).toHaveLength(1) + expect(queue[0]?.content).toBe('second') + expect(onQueueChange).toHaveBeenCalledWith( + expect.arrayContaining([expect.objectContaining({ content: 'second' })]), + ) + expect(onQueueChange.mock.calls.at(-1)?.[0]).toHaveLength(1) + + release() + await firstSend + }) + + it('cancelQueued removes a queued item before it drains', async () => { + const { connection, release } = createHoldingConnection() + const client = new ChatClient({ connection }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + await client.sendMessage('second') + const queued = client.getQueue() + expect(queued).toHaveLength(1) + const queuedId = queued[0]!.id + + client.cancelQueued(queuedId) + expect(client.getQueue()).toEqual([]) + + release() + await firstSend + }) + + it("whenBusy: 'drop' ignores a mid-stream send", async () => { + const { connection, release } = createHoldingConnection() + const client = new ChatClient({ connection, queue: 'drop' }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + await client.sendMessage('second') + expect(client.getQueue()).toEqual([]) + + release() + await firstSend + }) +}) From 62e7d5cb73b6d670e8e96c88cf055692bde114a8 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:11:49 +0200 Subject: [PATCH 05/23] feat(ai-client): auto-drain queue (fifo/batch) + flush on stop/clear/unsubscribe Wires the queue drain into streamResponse's settle path so queued sends actually go out (fifo one-at-a-time, or batch merged with newlines), and flushes any pending queue on stop()/clear()/unsubscribe() so callers don't get a surprise send after tearing down. Also surfaces the queue on the devtools snapshot. --- packages/ai-client/src/chat-client.ts | 79 +++++++++++++- packages/ai-client/src/devtools.ts | 2 + .../ai-client/tests/chat-client-queue.test.ts | 103 ++++++++++++++++++ packages/ai-client/tests/chat-client.test.ts | 11 +- 4 files changed, 186 insertions(+), 9 deletions(-) diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index c198cfb66..9879d3dc5 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -121,6 +121,38 @@ export function normalizeQueueOption( } } +/** + * Merge a run of queued messages into a single send for `drain: 'batch'`. + * All-string content is joined with newlines; mixed/multimodal content is + * flattened into a single `ContentPart` array. The last item's `body` wins. + */ +function mergeQueuedMessages( + items: Array }>, +): { content: string | MultimodalContent; body?: Record } { + const body = items.at(-1)?.body + const allString = items.every((i) => typeof i.content === 'string') + if (allString) { + return { + content: items.map((i) => i.content as string).join('\n'), + ...(body !== undefined ? { body } : {}), + } + } + const parts: Array = [] + for (const item of items) { + if (typeof item.content === 'string') { + parts.push({ type: 'text', content: item.content }) + } else if (typeof item.content.content === 'string') { + parts.push({ type: 'text', content: item.content.content }) + } else { + parts.push(...item.content.content) + } + } + return { + content: { content: parts }, + ...(body !== undefined ? { body } : {}), + } +} + export class ChatClient< TTools extends ReadonlyArray = any, TContext = unknown, @@ -609,6 +641,7 @@ export class ChatClient< connectionStatus: this.connectionStatus, sessionGenerating: this.sessionGenerating, activeRunIds: Array.from(this.activeRunIds), + queue: this.getQueue(), ...(this.error ? { error: this.error.message } : {}), } } @@ -1169,11 +1202,18 @@ export class ChatClient< } catch (error) { console.error('Failed to continue flow after tool result:', error) } - } else if (this.status !== 'ready') { - // Terminal run, but onStreamEnd never fired: the processor had no - // assistant message to emit it for (e.g. a bare RUN_FINISHED{stop}, - // #421). The normal path already set 'ready', so this is a no-op. - this.setStatus('ready') + } else { + if (this.status !== 'ready') { + // Terminal run, but onStreamEnd never fired: the processor had + // no assistant message to emit it for (e.g. a bare + // RUN_FINISHED{stop}, #421). The normal path already set + // 'ready', so this is a no-op. + this.setStatus('ready') + } + // Auto-send queued messages once the run fully settles. When a + // continuation runs instead (tool-result branch above), that + // continuation's own finally drains the queue. + await this.drainQueue() } } } @@ -1210,6 +1250,7 @@ export class ChatClient< setReadyStatus: true, abortSubscription: true, }) + this.flushQueue() this.resetSessionGenerating() this.setIsSubscribed(false) this.setConnectionStatus('disconnected') @@ -1250,6 +1291,7 @@ export class ChatClient< stop(): void { const hadLocalStream = this.abortController !== null this.cancelInFlightStream({ setReadyStatus: true }) + this.flushQueue() if (hadLocalStream) { this.resetSessionGenerating() } @@ -1277,6 +1319,7 @@ export class ChatClient< this.persistor.beginClear() } this.processor.clearMessages() + this.flushQueue() this.persistor?.remove() this.setError(undefined) this.events.messagesCleared() @@ -1472,6 +1515,30 @@ export class ChatClient< return this.processor.getMessages() as Array> } + /** + * Send the next queued message (fifo) or all queued messages merged into one + * (batch). Called from streamResponse's finally once the run has settled; + * the fifo case relies on the resulting stream's own finally to continue the + * chain. + */ + private async drainQueue(): Promise { + if (this.isLoading || this.messageQueue.length === 0) { + return + } + if (this.queueConfig.drain === 'batch') { + const items = this.messageQueue.splice(0) + this.emitQueueChange() + const merged = mergeQueuedMessages(items) + await this.sendMessage(merged.content, merged.body) + return + } + const next = this.messageQueue.shift() + this.emitQueueChange() + if (next) { + await this.sendMessage(next.content, next.body) + } + } + /** * Get the current send queue (messages held while a stream was in flight). */ @@ -1498,7 +1565,7 @@ export class ChatClient< this.emitQueueChange() } - protected flushQueue(): void { + private flushQueue(): void { if (this.messageQueue.length === 0) return this.messageQueue = [] this.emitQueueChange() diff --git a/packages/ai-client/src/devtools.ts b/packages/ai-client/src/devtools.ts index 07228ed73..a663b110e 100644 --- a/packages/ai-client/src/devtools.ts +++ b/packages/ai-client/src/devtools.ts @@ -16,6 +16,7 @@ import type { ChatClientState, ConnectionStatus, MessagePart, + QueuedMessage, ToolCallPart, UIMessage, } from './types' @@ -114,6 +115,7 @@ export interface AIDevtoolsChatSnapshot { connectionStatus: ConnectionStatus sessionGenerating: boolean activeRunIds: Array + queue?: Array error?: string } diff --git a/packages/ai-client/tests/chat-client-queue.test.ts b/packages/ai-client/tests/chat-client-queue.test.ts index 52f46467c..64090fdd7 100644 --- a/packages/ai-client/tests/chat-client-queue.test.ts +++ b/packages/ai-client/tests/chat-client-queue.test.ts @@ -128,3 +128,106 @@ describe('ChatClient message queue', () => { await firstSend }) }) + +/** + * Like {@link createHoldingConnection}, but only the first `connect()` call + * waits on the deferred; every subsequent call (i.e. a drained queue item's + * own send) resolves immediately with a unique messageId so chained drains + * don't collide on the same assistant message id. + */ +function createSequencedHoldingConnection(): { + connection: ConnectConnectionAdapter + release: () => void +} { + const deferred = createDeferred() + let call = 0 + const connection: ConnectConnectionAdapter = { + async *connect() { + call += 1 + if (call === 1) { + await deferred.promise + } + yield* createTextChunks('done', `msg-${call}`) + }, + } + return { connection, release: () => deferred.resolve() } +} + +describe('ChatClient queue drain', () => { + it('drains FIFO after the stream settles, in order', async () => { + const { connection, release } = createSequencedHoldingConnection() + const client = new ChatClient({ connection }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + await client.sendMessage('second') + await client.sendMessage('third') + expect(client.getQueue().map((m) => m.content)).toEqual([ + 'second', + 'third', + ]) + + release() + await firstSend + + expect(client.getQueue()).toEqual([]) + const userMessages = client + .getMessages() + .filter((m) => m.role === 'user') + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'first' }, + { type: 'text', content: 'second' }, + { type: 'text', content: 'third' }, + ]) + }) + + it('batch drain merges string queued items with newlines', async () => { + const { connection, release } = createSequencedHoldingConnection() + const client = new ChatClient({ connection, queue: { drain: 'batch' } }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + await client.sendMessage('a') + await client.sendMessage('b') + expect(client.getQueue().map((m) => m.content)).toEqual(['a', 'b']) + + release() + await firstSend + + expect(client.getQueue()).toEqual([]) + const userMessages = client + .getMessages() + .filter((m) => m.role === 'user') + expect(userMessages).toHaveLength(2) + expect(userMessages[1]?.parts[0]).toEqual({ + type: 'text', + content: 'a\nb', + }) + }) + + it('flushes the queue on stop()', async () => { + const { connection, release } = createHoldingConnection() + const client = new ChatClient({ connection }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + await client.sendMessage('second') + expect(client.getQueue()).toHaveLength(1) + + client.stop() + expect(client.getQueue()).toEqual([]) + + // Let the held-open stream settle so it doesn't leak into other tests. + release() + await firstSend + }) +}) diff --git a/packages/ai-client/tests/chat-client.test.ts b/packages/ai-client/tests/chat-client.test.ts index 7f8cb66c4..d329928c2 100644 --- a/packages/ai-client/tests/chat-client.test.ts +++ b/packages/ai-client/tests/chat-client.test.ts @@ -2137,7 +2137,7 @@ describe('ChatClient', () => { expect(client.getMessages().length).toBe(0) }) - it('should not send message while loading', async () => { + it('should queue (not send immediately) a message sent while loading, then auto-send it once the stream settles', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 100, @@ -2149,9 +2149,14 @@ describe('ChatClient', () => { await Promise.all([promise1, promise2]) - // Should only have one user message since second was blocked + // The second send is queued (default `whenBusy: 'queue'`) rather than + // started concurrently, then auto-drains once the first stream settles + // — both end up sent, in order. const userMessages = client.getMessages().filter((m) => m.role === 'user') - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) }) From ac8ef62f8b863e2d328b226d67e3988307d9c488 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:21:53 +0200 Subject: [PATCH 06/23] fix(ai-client): flush message queue when a stream errors A RUN_ERROR or thrown non-abort error left streamCompletedSuccessfully false, so the finally block's queue handling (guarded by that flag) was skipped entirely. Queued messages were stranded until a later direct sendMessage sent first and drained the stale queued item afterward, inverting message order. Flush (not drain) the queue on the non-success settle path, matching stop()'s existing behavior. --- packages/ai-client/src/chat-client.ts | 6 ++ .../ai-client/tests/chat-client-queue.test.ts | 61 +++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index 9879d3dc5..5e4e593bf 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -1215,6 +1215,12 @@ export class ChatClient< // continuation's own finally drains the queue. await this.drainQueue() } + } else { + // Error/abort settle for the active generation: don't strand or + // later mis-order queued messages. A failed turn flushes the queue + // (consistent with stop()); it must NOT auto-drain into a likely + // broken endpoint. + this.flushQueue() } } } diff --git a/packages/ai-client/tests/chat-client-queue.test.ts b/packages/ai-client/tests/chat-client-queue.test.ts index 64090fdd7..ab3dfbf39 100644 --- a/packages/ai-client/tests/chat-client-queue.test.ts +++ b/packages/ai-client/tests/chat-client-queue.test.ts @@ -1,7 +1,9 @@ import { describe, expect, it, vi } from 'vitest' +import { EventType } from '@tanstack/ai/client' import { ChatClient, normalizeQueueOption } from '../src/chat-client' import { createTextChunks } from './test-utils' import type { ConnectConnectionAdapter } from '../src/connection-adapters' +import type { StreamChunk } from '@tanstack/ai/client' describe('normalizeQueueOption', () => { it('defaults to queue + fifo + reject', () => { @@ -126,6 +128,15 @@ describe('ChatClient message queue', () => { release() await firstSend + + // The dropped message must never have become a real message either — + // it should be gone entirely, not just missing from the queue. + const userMessages = client + .getMessages() + .filter((m) => m.role === 'user') + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'first' }, + ]) }) }) @@ -153,6 +164,30 @@ function createSequencedHoldingConnection(): { return { connection, release: () => deferred.resolve() } } +/** + * Like {@link createHoldingConnection}, but `release()` settles the stream + * with a `RUN_ERROR` chunk instead of a successful text response. + */ +function createErroringHoldingConnection(): { + connection: ConnectConnectionAdapter + release: () => void +} { + const deferred = createDeferred() + const connection: ConnectConnectionAdapter = { + async *connect(): AsyncGenerator { + await deferred.promise + yield { + type: EventType.RUN_ERROR, + threadId: 'thread-1', + timestamp: Date.now(), + message: 'boom', + error: { message: 'boom' }, + } as StreamChunk + }, + } + return { connection, release: () => deferred.resolve() } +} + describe('ChatClient queue drain', () => { it('drains FIFO after the stream settles, in order', async () => { const { connection, release } = createSequencedHoldingConnection() @@ -230,4 +265,30 @@ describe('ChatClient queue drain', () => { release() await firstSend }) + + it('flushes the queue when the stream errors', async () => { + const { connection, release } = createErroringHoldingConnection() + const client = new ChatClient({ connection }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + await client.sendMessage('second') + expect(client.getQueue()).toHaveLength(1) + + release() + await firstSend + + // The queue must be flushed, not stranded or auto-drained into the + // now-broken endpoint. + expect(client.getQueue()).toEqual([]) + const userMessages = client + .getMessages() + .filter((m) => m.role === 'user') + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'first' }, + ]) + }) }) From 2c4bb89048b44a6f7256f62ef6c77edfe73658bb Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:26:06 +0200 Subject: [PATCH 07/23] feat(ai-svelte): expose queue + cancelQueued + per-send whenBusy --- packages/ai-svelte/src/create-chat.svelte.ts | 20 +++++++++++++-- packages/ai-svelte/src/types.ts | 26 ++++++++++++++++++-- 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/packages/ai-svelte/src/create-chat.svelte.ts b/packages/ai-svelte/src/create-chat.svelte.ts index 506d9843f..02b701b60 100644 --- a/packages/ai-svelte/src/create-chat.svelte.ts +++ b/packages/ai-svelte/src/create-chat.svelte.ts @@ -4,7 +4,9 @@ import type { ChatClientState, ConnectionStatus, InferredClientContext, + QueuedMessage, StructuredOutputPart, + WhenBusy, } from '@tanstack/ai-client' import type { AnyClientTool, @@ -71,6 +73,7 @@ export function createChat< let isSubscribed = $state(false) let connectionStatus = $state('disconnected') let sessionGenerating = $state(false) + let queue = $state>([]) // Structured-output `partial` / `final` are derived from `messages` — // specifically from the structured-output part on the latest assistant @@ -153,6 +156,10 @@ export function createChat< onSessionGeneratingChange: (isGenerating: boolean) => { sessionGenerating = isGenerating }, + ...(options.queue !== undefined && { queue: options.queue }), + onQueueChange: (nextQueue: Array) => { + queue = nextQueue + }, }) messages = client.getMessages() @@ -169,10 +176,15 @@ export function createChat< // Users should call chat.stop() in their component's cleanup if needed. // Define methods - const sendMessage = async (content: string | MultimodalContent) => { - await client.sendMessage(content) + const sendMessage = async ( + content: string | MultimodalContent, + sendOptions?: { whenBusy?: WhenBusy }, + ) => { + await client.sendMessage(content, undefined, sendOptions) } + const cancelQueued = (id: string) => client.cancelQueued(id) + const append = async (message: ModelMessage | UIMessage) => { await client.append(message) } @@ -293,6 +305,9 @@ export function createChat< get sessionGenerating() { return sessionGenerating }, + get queue() { + return queue + }, get partial() { return partial }, @@ -300,6 +315,7 @@ export function createChat< return final }, sendMessage, + cancelQueued, append, reload, stop, diff --git a/packages/ai-svelte/src/types.ts b/packages/ai-svelte/src/types.ts index ffa7c0122..f7bf3ae83 100644 --- a/packages/ai-svelte/src/types.ts +++ b/packages/ai-svelte/src/types.ts @@ -14,11 +14,19 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueuedMessage, UIMessage, + WhenBusy, } from '@tanstack/ai-client' // Re-export types from ai-client -export type { ChatRequestBody, MultimodalContent, UIMessage } +export type { + ChatRequestBody, + MultimodalContent, + QueuedMessage, + UIMessage, + WhenBusy, +} /** * Recursive partial — every property and every nested array element is @@ -66,6 +74,7 @@ export type CreateChatOptions< | 'onSubscriptionChange' | 'onConnectionStatusChange' | 'onSessionGeneratingChange' + | 'onQueueChange' | 'context' | 'devtools' > & { @@ -126,7 +135,20 @@ interface BaseCreateChatReturn< * Send a message and get a response. * Can be a simple string or multimodal content with images, audio, etc. */ - sendMessage: (content: string | MultimodalContent) => Promise + sendMessage: ( + content: string | MultimodalContent, + options?: { whenBusy?: WhenBusy }, + ) => Promise + + /** + * Pending messages queued while a stream is in flight. + */ + queue: Array + + /** + * Cancel a queued message before it drains. No-op if already sent. + */ + cancelQueued: (id: string) => void /** * Append a message to the conversation From 833e7127f4fb2103dc0a5749bef0405921df67df Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:26:17 +0200 Subject: [PATCH 08/23] feat(ai-vue): expose queue + cancelQueued + per-send whenBusy --- packages/ai-vue/src/types.ts | 20 ++++++++++++++++++-- packages/ai-vue/src/use-chat.ts | 18 ++++++++++++++++-- 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/packages/ai-vue/src/types.ts b/packages/ai-vue/src/types.ts index 60fe898fd..faaf3b044 100644 --- a/packages/ai-vue/src/types.ts +++ b/packages/ai-vue/src/types.ts @@ -14,12 +14,14 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueuedMessage, UIMessage, + WhenBusy, } from '@tanstack/ai-client' import type { DeepReadonly, ShallowRef } from 'vue' // Re-export types from ai-client -export type { ChatRequestBody, MultimodalContent, UIMessage } +export type { ChatRequestBody, MultimodalContent, QueuedMessage, UIMessage, WhenBusy } /** * Recursive partial — every property and every nested array element is optional. @@ -68,6 +70,7 @@ export type UseChatOptions< | 'onSubscriptionChange' | 'onConnectionStatusChange' | 'onSessionGeneratingChange' + | 'onQueueChange' | 'context' | 'devtools' > & { @@ -124,7 +127,20 @@ interface BaseUseChatReturn< * Send a message and get a response. * Can be a simple string or multimodal content with images, audio, etc. */ - sendMessage: (content: string | MultimodalContent) => Promise + sendMessage: ( + content: string | MultimodalContent, + options?: { whenBusy?: WhenBusy }, + ) => Promise + + /** + * Pending messages queued while a stream is in flight. + */ + queue: Readonly>> + + /** + * Cancel a queued message before it drains. No-op if already sent. + */ + cancelQueued: (id: string) => void /** * Append a message to the conversation diff --git a/packages/ai-vue/src/use-chat.ts b/packages/ai-vue/src/use-chat.ts index 31719591d..79245485b 100644 --- a/packages/ai-vue/src/use-chat.ts +++ b/packages/ai-vue/src/use-chat.ts @@ -20,7 +20,9 @@ import type { ChatClientState, ConnectionStatus, InferredClientContext, + QueuedMessage, StructuredOutputPart, + WhenBusy, } from '@tanstack/ai-client' import type { DeepPartial, @@ -53,6 +55,7 @@ export function useChat< const isSubscribed = shallowRef(false) const connectionStatus = shallowRef('disconnected') const sessionGenerating = shallowRef(false) + const queue = shallowRef>([]) // Structured-output `partial` / `final` are derived from `messages` — // specifically from the structured-output part on the latest assistant @@ -138,6 +141,10 @@ export function useChat< onSessionGeneratingChange: (isGenerating: boolean) => { sessionGenerating.value = isGenerating }, + ...(options.queue !== undefined && { queue: options.queue }), + onQueueChange: (nextQueue: Array) => { + queue.value = nextQueue + }, }) messages.value = client.getMessages() @@ -190,10 +197,15 @@ export function useChat< // Callback options are read through `options.xxx` at call time, so reactive // or mutated options propagate without recreating the client. - const sendMessage = async (content: string | MultimodalContent) => { - await client.sendMessage(content) + const sendMessage = async ( + content: string | MultimodalContent, + sendOptions?: { whenBusy?: WhenBusy }, + ) => { + await client.sendMessage(content, undefined, sendOptions) } + const cancelQueued = (id: string) => client.cancelQueued(id) + const append = async (message: ModelMessage | UIMessage) => { await client.append(message) } @@ -279,6 +291,8 @@ export function useChat< return { messages: readonly(messages), sendMessage, + queue: readonly(queue), + cancelQueued, append, reload, stop, From 17bb417d30ff7e339878a17fdf47bcefba302660 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:27:49 +0200 Subject: [PATCH 09/23] feat(ai-react): expose queue + cancelQueued + per-send whenBusy --- packages/ai-react/src/types.ts | 26 ++++++++++++++++++++++++-- packages/ai-react/src/use-chat.ts | 26 ++++++++++++++++++++++++-- 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/packages/ai-react/src/types.ts b/packages/ai-react/src/types.ts index 87c4076f1..429b08262 100644 --- a/packages/ai-react/src/types.ts +++ b/packages/ai-react/src/types.ts @@ -14,11 +14,19 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueuedMessage, UIMessage, + WhenBusy, } from '@tanstack/ai-client' // Re-export types from ai-client -export type { ChatRequestBody, MultimodalContent, UIMessage } +export type { + ChatRequestBody, + MultimodalContent, + QueuedMessage, + UIMessage, + WhenBusy, +} /** * Recursive partial — every property and every nested array element is optional. @@ -71,6 +79,7 @@ export type UseChatOptions< | 'onSubscriptionChange' | 'onConnectionStatusChange' | 'onSessionGeneratingChange' + | 'onQueueChange' | 'context' | 'devtools' > & { @@ -135,7 +144,20 @@ interface BaseUseChatReturn< * Send a message and get a response. * Can be a simple string or multimodal content with images, audio, etc. */ - sendMessage: (content: string | MultimodalContent) => Promise + sendMessage: ( + content: string | MultimodalContent, + options?: { whenBusy?: WhenBusy }, + ) => Promise + + /** + * Pending messages queued while a stream is in flight. + */ + queue: Array + + /** + * Cancel a queued message before it drains. No-op if already sent. + */ + cancelQueued: (id: string) => void /** * Append a message to the conversation diff --git a/packages/ai-react/src/use-chat.ts b/packages/ai-react/src/use-chat.ts index 37444b569..f2f596e44 100644 --- a/packages/ai-react/src/use-chat.ts +++ b/packages/ai-react/src/use-chat.ts @@ -12,7 +12,9 @@ import type { ChatClientState, ConnectionStatus, InferredClientContext, + QueuedMessage, StructuredOutputPart, + WhenBusy, } from '@tanstack/ai-client' import type { @@ -43,6 +45,7 @@ export function useChat< const [connectionStatus, setConnectionStatus] = useState('disconnected') const [sessionGenerating, setSessionGenerating] = useState(false) + const [queue, setQueue] = useState>([]) type Partial = DeepPartial>> type Final = InferSchemaType> @@ -156,6 +159,13 @@ export function useChat< if (activeClientRef.current !== instance) return setSessionGenerating(isGenerating) }, + ...(optionsRef.current.queue !== undefined && { + queue: optionsRef.current.queue, + }), + onQueueChange: (nextQueue: Array) => { + if (activeClientRef.current !== instance) return + setQueue(nextQueue) + }, }) activeClientRef.current = instance return instance @@ -229,8 +239,18 @@ export function useChat< }, [client]) const sendMessage = useCallback( - async (content: string | MultimodalContent) => { - await client.sendMessage(content) + async ( + content: string | MultimodalContent, + sendOptions?: { whenBusy?: WhenBusy }, + ) => { + await client.sendMessage(content, undefined, sendOptions) + }, + [client], + ) + + const cancelQueued = useCallback( + (id: string) => { + client.cancelQueued(id) }, [client], ) @@ -346,6 +366,8 @@ export function useChat< clear, addToolResult, addToolApprovalResponse, + queue, + cancelQueued, partial, final, } as unknown as UseChatReturn From 67d16a2f519b291822de0f39980f3c781e1001fb Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:29:21 +0200 Subject: [PATCH 10/23] feat(ai-solid): expose queue + cancelQueued + per-send whenBusy --- packages/ai-solid/src/types.ts | 27 +++++++++++++++++++++++++-- packages/ai-solid/src/use-chat.ts | 18 ++++++++++++++++-- 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/packages/ai-solid/src/types.ts b/packages/ai-solid/src/types.ts index cc7d99a3b..4dc532d9d 100644 --- a/packages/ai-solid/src/types.ts +++ b/packages/ai-solid/src/types.ts @@ -14,12 +14,20 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueuedMessage, UIMessage, + WhenBusy, } from '@tanstack/ai-client' import type { Accessor } from 'solid-js' // Re-export types from ai-client -export type { ChatRequestBody, MultimodalContent, UIMessage } +export type { + ChatRequestBody, + MultimodalContent, + QueuedMessage, + UIMessage, + WhenBusy, +} /** * Recursive partial — every property and every nested array element is optional. @@ -67,6 +75,7 @@ export type UseChatOptions< | 'onSubscriptionChange' | 'onConnectionStatusChange' | 'onSessionGeneratingChange' + | 'onQueueChange' | 'context' | 'devtools' > & { @@ -117,11 +126,25 @@ interface BaseUseChatReturn< */ messages: Accessor>> + /** + * Messages currently queued because a request was in flight when they + * were sent (see `sendMessage`'s `whenBusy: 'enqueue'` option). + */ + queue: Accessor> + + /** + * Cancel a previously queued message by id. + */ + cancelQueued: (id: string) => void + /** * Send a message and get a response. * Can be a simple string or multimodal content with images, audio, etc. */ - sendMessage: (content: string | MultimodalContent) => Promise + sendMessage: ( + content: string | MultimodalContent, + options?: { whenBusy?: WhenBusy }, + ) => Promise /** * Append a message to the conversation diff --git a/packages/ai-solid/src/use-chat.ts b/packages/ai-solid/src/use-chat.ts index 9a6e404cf..982e60bad 100644 --- a/packages/ai-solid/src/use-chat.ts +++ b/packages/ai-solid/src/use-chat.ts @@ -13,7 +13,9 @@ import type { ChatClientState, ConnectionStatus, InferredClientContext, + QueuedMessage, StructuredOutputPart, + WhenBusy, } from '@tanstack/ai-client' import type { AnyClientTool, @@ -54,6 +56,7 @@ export function useChat< const [connectionStatus, setConnectionStatus] = createSignal('disconnected') const [sessionGenerating, setSessionGenerating] = createSignal(false) + const [queue, setQueue] = createSignal>([]) // Structured-output `partial` / `final` are derived from `messages` — // specifically from the structured-output part on the latest assistant @@ -135,6 +138,10 @@ export function useChat< onSessionGeneratingChange: (isGenerating: boolean) => { setSessionGenerating(isGenerating) }, + ...(options.queue !== undefined && { queue: options.queue }), + onQueueChange: (nextQueue: Array) => { + setQueue(nextQueue) + }, }) // Only recreate when clientId changes // Connection and other options are captured at creation time @@ -189,8 +196,11 @@ export function useChat< // Callback options are read through `options.xxx` at call time, so reactive // or mutated options propagate without recreating the client. - const sendMessage = async (content: string | MultimodalContent) => { - await client().sendMessage(content) + const sendMessage = async ( + content: string | MultimodalContent, + sendOptions?: { whenBusy?: WhenBusy }, + ) => { + await client().sendMessage(content, undefined, sendOptions) } const append = async (message: ModelMessage | UIMessage) => { @@ -230,6 +240,8 @@ export function useChat< await client().addToolApprovalResponse(response) } + const cancelQueued = (id: string) => client().cancelQueued(id) + // The "active" structured-output part is on the assistant message after // the latest user message. When no user message exists yet, return null // rather than scanning history — otherwise a stale `final` from @@ -273,6 +285,7 @@ export function useChat< // eslint-disable-next-line no-restricted-syntax -- primitive return shape diverges from generic UseChatReturn; TS can't structurally narrow the conditional partial/final fields return { messages, + queue, sendMessage, append, reload, @@ -287,6 +300,7 @@ export function useChat< clear, addToolResult, addToolApprovalResponse, + cancelQueued, partial, final, } as unknown as UseChatReturn From 31a01c1be5401ba6c1d6fcfada1c8ebaad73b173 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:37:20 +0200 Subject: [PATCH 11/23] test: update framework hook tests for queue-by-default behavior --- packages/ai-react/tests/use-chat.test.ts | 21 +++++++++++++++------ packages/ai-solid/tests/use-chat.test.ts | 21 +++++++++++++++------ packages/ai-vue/tests/use-chat.test.ts | 21 +++++++++++++++------ 3 files changed, 45 insertions(+), 18 deletions(-) diff --git a/packages/ai-react/tests/use-chat.test.ts b/packages/ai-react/tests/use-chat.test.ts index 8e43eedc1..249306e7a 100644 --- a/packages/ai-react/tests/use-chat.test.ts +++ b/packages/ai-react/tests/use-chat.test.ts @@ -375,7 +375,7 @@ describe('useChat', () => { expect(result.current.messages.length).toBe(0) }) - it('should not send message while loading', async () => { + it('should queue a message sent while loading and send it after', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 100, @@ -387,11 +387,16 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) - // Should only have one user message since second was blocked + // The second send is queued (default `whenBusy: 'queue'`) while the + // first stream is in flight, then auto-drains once it settles — both + // end up sent, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle errors during sendMessage', async () => { @@ -1189,7 +1194,7 @@ describe('useChat', () => { }) describe('concurrent operations', () => { - it('should handle multiple sendMessage calls', async () => { + it('should queue and then deliver multiple sendMessage calls in order', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 50, @@ -1201,11 +1206,15 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) - // Should only have one user message (second should be blocked) + // The second call is queued while the first stream is in flight, + // then auto-sent once it settles — both land, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle stop during sendMessage', async () => { diff --git a/packages/ai-solid/tests/use-chat.test.ts b/packages/ai-solid/tests/use-chat.test.ts index 31ba2f26f..c8eb4f35f 100644 --- a/packages/ai-solid/tests/use-chat.test.ts +++ b/packages/ai-solid/tests/use-chat.test.ts @@ -308,7 +308,7 @@ describe('useChat', () => { expect(result.current.messages.length).toBe(0) }) - it('should not send message while loading', async () => { + it('should queue a message sent while loading and send it after', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 100, @@ -320,11 +320,16 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) - // Should only have one user message since second was blocked + // The second send is queued (default `whenBusy: 'queue'`) while the + // first stream is in flight, then auto-drains once it settles — both + // end up sent, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle errors during sendMessage', async () => { @@ -927,7 +932,7 @@ describe('useChat', () => { }) describe('concurrent operations', () => { - it('should handle multiple sendMessage calls', async () => { + it('should queue and then deliver multiple sendMessage calls in order', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 50, @@ -939,11 +944,15 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) - // Should only have one user message (second should be blocked) + // The second call is queued while the first stream is in flight, + // then auto-sent once it settles — both land, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle stop during sendMessage', async () => { diff --git a/packages/ai-vue/tests/use-chat.test.ts b/packages/ai-vue/tests/use-chat.test.ts index 0e227afe5..9ed8b77e6 100644 --- a/packages/ai-vue/tests/use-chat.test.ts +++ b/packages/ai-vue/tests/use-chat.test.ts @@ -289,7 +289,7 @@ describe('useChat', () => { expect(result.current.messages.length).toBe(0) }) - it('should not send message while loading', async () => { + it('should queue a message sent while loading and send it after', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 100, @@ -302,11 +302,16 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) await flushPromises() - // Should only have one user message since second was blocked + // The second send is queued (default `whenBusy: 'queue'`) while the + // first stream is in flight, then auto-drains once it settles — both + // end up sent, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle errors during sendMessage', async () => { @@ -872,7 +877,7 @@ describe('useChat', () => { }) describe('concurrent operations', () => { - it('should handle multiple sendMessage calls', async () => { + it('should queue and then deliver multiple sendMessage calls in order', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 50, @@ -885,11 +890,15 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) await flushPromises() - // Should only have one user message (second should be blocked) + // The second call is queued while the first stream is in flight, + // then auto-sent once it settles — both land, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle stop during sendMessage', async () => { From 7e6a99e60b04d35c321ed131c5659c1fc6795862 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:39:01 +0200 Subject: [PATCH 12/23] docs: add changeset for client message queue --- .changeset/message-queue.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .changeset/message-queue.md diff --git a/.changeset/message-queue.md b/.changeset/message-queue.md new file mode 100644 index 000000000..d2f440907 --- /dev/null +++ b/.changeset/message-queue.md @@ -0,0 +1,13 @@ +--- +'@tanstack/ai-client': minor +'@tanstack/ai-react': minor +'@tanstack/ai-solid': minor +'@tanstack/ai-vue': minor +'@tanstack/ai-svelte': minor +--- + +Messages sent while a stream is already in flight are now queued by default and automatically sent once the in-flight stream settles, instead of being silently dropped. **This is a behavior change.** + +The behavior is configurable via a new `queue` option, which accepts `whenBusy: 'queue' | 'drop' | 'interrupt'`, `drain: 'fifo' | 'batch'`, `maxSize`, and `onOverflow`, or a custom strategy function for full control. + +Queued messages are exposed on the hook as `queue` and can be cancelled before they send via `cancelQueued(id)`. `sendMessage` also accepts a per-call `{ whenBusy }` override. From 7b265068cf6e052b7fcad8d22d2909aafbe25501 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:41:14 +0200 Subject: [PATCH 13/23] docs(skills): document message queue in chat-experience skill --- .../skills/ai-core/chat-experience/SKILL.md | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/packages/ai/skills/ai-core/chat-experience/SKILL.md b/packages/ai/skills/ai-core/chat-experience/SKILL.md index 3ae4408dc..2a0129f8f 100644 --- a/packages/ai/skills/ai-core/chat-experience/SKILL.md +++ b/packages/ai/skills/ai-core/chat-experience/SKILL.md @@ -409,6 +409,53 @@ export const Route = createFileRoute('/api/chat')({ }) ``` +### 7. Queueing Messages Sent While Streaming + +By default, a `sendMessage` call that arrives while a stream is in flight is +**queued** and sent automatically once the stream settles — this is a +behavior change: such sends used to be silently dropped. Configure it with +the `queue` option on `useChat`: + +```typescript +import { useChat, fetchServerSentEvents } from '@tanstack/ai-react' + +const { messages, queue, sendMessage, cancelQueued, isLoading } = useChat({ + connection: fetchServerSentEvents('/api/chat'), + queue: { whenBusy: 'queue', drain: 'fifo', maxSize: 5, onOverflow: 'reject' }, +}) +``` + +- **`whenBusy`** — `'queue'` (default) holds the message; `'drop'` ignores + the send; `'interrupt'` aborts the current stream (like `stop()`) and + sends immediately. Also accepts a plain `WhenBusy` string as shorthand, or + a `QueueStrategy` function for full per-send control. +- **`drain`** — `'fifo'` (default) sends queued items one at a time in + order; `'batch'` merges everything queued into a single send once the + stream settles. +- **`maxSize`** / **`onOverflow`** — cap the queue length; `'reject'` + (default) ignores overflow sends, `'drop-oldest'` evicts the oldest + queued item to make room. + +`queue: Array` (`{ id, content, createdAt }`) is separate +from `messages` — render pending sends distinctly and cancel with +`cancelQueued(id)`: + +```typescript +{queue.map((q) => ( +
+ {typeof q.content === 'string' ? q.content : '[attachment]'} + +
+))} +``` + +Override the configured policy for a single send with the second argument +to `sendMessage`: + +```typescript +sendMessage('Never mind, do this instead', { whenBusy: 'interrupt' }) +``` + ## Common Mistakes ### a. CRITICAL: Using Vercel AI SDK patterns (streamText, generateText) From 7b1986809312ecf4c8dc12e0b23da1d6b88fa958 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:44:47 +0200 Subject: [PATCH 14/23] fix(ai-vue,ai-solid): reflow re-export + correct queue jsdoc --- packages/ai-solid/src/types.ts | 3 +-- packages/ai-vue/src/types.ts | 8 +++++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/packages/ai-solid/src/types.ts b/packages/ai-solid/src/types.ts index 4dc532d9d..c3466a096 100644 --- a/packages/ai-solid/src/types.ts +++ b/packages/ai-solid/src/types.ts @@ -127,8 +127,7 @@ interface BaseUseChatReturn< messages: Accessor>> /** - * Messages currently queued because a request was in flight when they - * were sent (see `sendMessage`'s `whenBusy: 'enqueue'` option). + * Pending messages queued while a stream is in flight. */ queue: Accessor> diff --git a/packages/ai-vue/src/types.ts b/packages/ai-vue/src/types.ts index faaf3b044..0235df297 100644 --- a/packages/ai-vue/src/types.ts +++ b/packages/ai-vue/src/types.ts @@ -21,7 +21,13 @@ import type { import type { DeepReadonly, ShallowRef } from 'vue' // Re-export types from ai-client -export type { ChatRequestBody, MultimodalContent, QueuedMessage, UIMessage, WhenBusy } +export type { + ChatRequestBody, + MultimodalContent, + QueuedMessage, + UIMessage, + WhenBusy, +} /** * Recursive partial — every property and every nested array element is optional. From b6376a7e5658aa6d0d276e0b41a1e0677fb7d01f Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 15:51:18 +0200 Subject: [PATCH 15/23] docs: document client message queue (queue option, cancelQueued) --- docs/chat/streaming.md | 49 ++++++++++++++++++++++++++++++++++++++++++ docs/config.json | 2 +- 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/docs/chat/streaming.md b/docs/chat/streaming.md index db32def3c..8382629bb 100644 --- a/docs/chat/streaming.md +++ b/docs/chat/streaming.md @@ -203,6 +203,54 @@ export async function POST(request: Request) { } ``` +## Queueing Messages + +By default, calling `sendMessage` while a stream is already in flight **queues** the message instead of dropping it — it sends automatically once the current run settles. Configure this with the `queue` option, which accepts a `QueueConfig` object, a plain shorthand string, or a strategy function: + +```tsx group=queueing-messages +import { useChat, fetchServerSentEvents } from "@tanstack/ai-react"; + +const { messages, queue, sendMessage, cancelQueued, isLoading } = useChat({ + connection: fetchServerSentEvents("/api/chat"), + queue: { whenBusy: "queue", drain: "fifo", maxSize: 5 }, +}); +``` + +- **`whenBusy`** — what happens to a send that arrives mid-stream: + - `"queue"` (default) — hold the message; it sends once the stream settles. + - `"drop"` — ignore the send. + - `"interrupt"` — abort the current stream (like calling `stop()`) and send the new message immediately. +- **`drain`** — how queued items leave the queue: `"fifo"` (default) sends them one at a time in order; `"batch"` merges everything currently queued into a single send once the stream settles (string contents joined with `\n`, multimodal content concatenated in order). +- **`maxSize`** — caps how many messages can be queued. +- **`onOverflow`** — `"reject"` (default) ignores a send once `maxSize` is reached; `"drop-oldest"` evicts the oldest queued item to make room. + +You can also pass a plain `WhenBusy` string as shorthand for `queue: "interrupt"`, or a `QueueStrategy` function for full control over the per-send decision (the drain order stays FIFO for the function form). + +`useChat` exposes the pending queue as `queue` so you can render it distinctly from `messages`, along with `cancelQueued(id)` to cancel an item before it sends: + +```tsx group=queueing-messages +function PendingQueue() { + return ( + <> + {queue.map((q) => ( +
+ {typeof q.content === "string" ? q.content : "[attachment]"} + +
+ ))} + + ); +} +``` + +Override the configured policy for a single send with the second argument to `sendMessage`: + +```tsx group=queueing-messages +sendMessage("Never mind, do this instead", { whenBusy: "interrupt" }); +``` + +> **Note:** This is a default-behavior change — messages sent while streaming used to be silently dropped. They are now queued unless you opt into `whenBusy: "drop"` or `"interrupt"`. + ## Best Practices 1. **Handle loading states** - Use `isLoading` to show loading indicators @@ -210,6 +258,7 @@ export async function POST(request: Request) { 3. **Cancel on unmount** - Clean up streams when components unmount 4. **Optimize rendering** - Batch updates if needed for performance 5. **Show progress** - Display partial content as it streams +6. **Render queued messages distinctly** - Use `queue` to show pending sends separately from `messages` ## Next Steps diff --git a/docs/config.json b/docs/config.json index c7b16c418..1625517ae 100644 --- a/docs/config.json +++ b/docs/config.json @@ -157,7 +157,7 @@ "label": "Streaming", "to": "chat/streaming", "addedAt": "2026-04-15", - "updatedAt": "2026-07-03" + "updatedAt": "2026-07-06" }, { "label": "Connection Adapters", From 105b7414277479fee177b92439ab341deeba3485 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 16:00:31 +0200 Subject: [PATCH 16/23] test(e2e): queue-while-streaming, cancel, and drain ordering --- testing/e2e/fixtures/queue/basic.json | 22 +++++ testing/e2e/src/components/ChatUI.tsx | 54 ++++++++++- testing/e2e/src/routes/$provider/$feature.tsx | 4 + testing/e2e/tests/queue.spec.ts | 94 +++++++++++++++++++ 4 files changed, 170 insertions(+), 4 deletions(-) create mode 100644 testing/e2e/fixtures/queue/basic.json create mode 100644 testing/e2e/tests/queue.spec.ts diff --git a/testing/e2e/fixtures/queue/basic.json b/testing/e2e/fixtures/queue/basic.json new file mode 100644 index 000000000..dbd5e746a --- /dev/null +++ b/testing/e2e/fixtures/queue/basic.json @@ -0,0 +1,22 @@ +{ + "fixtures": [ + { + "match": { + "userMessage": "[queue] first message" + }, + "response": { + "content": "This is the first response and it keeps streaming for a good while so that the follow-up messages sent during this in-flight run land in the client queue instead of starting a second concurrent request right away, giving the test plenty of time to send more messages and cancel one before this run settles." + }, + "chunkSize": 2, + "streamingProfile": { "tps": 6 } + }, + { + "match": { + "userMessage": "[queue] third message" + }, + "response": { + "content": "This is the third response, delivered after the queue drained." + } + } + ] +} diff --git a/testing/e2e/src/components/ChatUI.tsx b/testing/e2e/src/components/ChatUI.tsx index 9bd218180..dd5c38591 100644 --- a/testing/e2e/src/components/ChatUI.tsx +++ b/testing/e2e/src/components/ChatUI.tsx @@ -1,9 +1,10 @@ import { useEffect, useRef, useState } from 'react' -import type { UIMessage } from '@tanstack/ai-react' import ReactMarkdown from 'react-markdown' import rehypeRaw from 'rehype-raw' import rehypeSanitize from 'rehype-sanitize' import remarkGfm from 'remark-gfm' +import type { UIMessage } from '@tanstack/ai-react' +import type { QueuedMessage } from '@tanstack/ai-client' import { ToolCallDisplay } from '@/components/ToolCallDisplay' import { ApprovalPrompt } from '@/components/ApprovalPrompt' @@ -25,6 +26,13 @@ interface ChatUIProps { /** Number of TEXT_MESSAGE_CONTENT chunks observed. Used by streaming e2e * tests to verify the response actually streamed in multiple deltas. */ contentDeltaCount?: number + /** Messages sent while a stream was already in flight — held here by + * `useChat` and auto-sent FIFO once the run settles. Rendered in a + * region separate from `messages` so e2e tests can assert queued state + * distinctly from the delivered conversation. */ + queue?: Array + /** Remove a queued message before it drains. */ + cancelQueued?: (id: string) => void } export function ChatUI({ @@ -37,6 +45,8 @@ export function ChatUI({ onStop, structuredObject, contentDeltaCount, + queue, + cancelQueued, }: ChatUIProps) { const [input, setInput] = useState('') const messagesRef = useRef(null) @@ -141,13 +151,13 @@ export function ChatUI({ return ( ) } if (part.type === 'tool-call') { - return + return } if (part.type === 'tool-result') { return ( @@ -195,6 +205,37 @@ export function ChatUI({ )} + {queue != null && queue.length > 0 && ( +
+ {queue.map((queued) => ( +
+ + {typeof queued.content === 'string' + ? queued.content + : JSON.stringify(queued.content)} + + {cancelQueued && ( + + )} +
+ ))} +
+ )} +
{showImageInput && ( Send diff --git a/testing/e2e/src/routes/$provider/$feature.tsx b/testing/e2e/src/routes/$provider/$feature.tsx index ed1dfa9ea..65df290a5 100644 --- a/testing/e2e/src/routes/$provider/$feature.tsx +++ b/testing/e2e/src/routes/$provider/$feature.tsx @@ -282,6 +282,8 @@ function ChatFeature({ addToolApprovalResponse, stop, clear, + queue, + cancelQueued, } = useChat({ id: chatId, ...transport, @@ -346,6 +348,8 @@ function ChatFeature({ isLoading={isLoading} structuredObject={structuredObject} contentDeltaCount={contentDeltaCount} + queue={queue} + cancelQueued={cancelQueued} onSendMessage={(text) => { sendMessage(text) }} diff --git a/testing/e2e/tests/queue.spec.ts b/testing/e2e/tests/queue.spec.ts new file mode 100644 index 000000000..05b59a86e --- /dev/null +++ b/testing/e2e/tests/queue.spec.ts @@ -0,0 +1,94 @@ +import { test, expect } from './fixtures' +import { sendMessage, waitForAssistantText, featureUrl } from './helpers' + +// Fixture: fixtures/queue/basic.json +// - "[queue] first message" streams slowly (tps: 6, chunkSize: 2) so the run +// stays in flight long enough to send follow-up messages while it streams. +// - "[queue] third message" resolves immediately. +// "[queue] second message" has no fixture on purpose — it's cancelled before +// it ever drains, so if a regression let it send anyway the request would +// fail to match and the test would fail loudly instead of silently passing. + +test.describe('client message queue', () => { + test('queues messages sent while streaming, supports cancel, and drains FIFO', async ({ + page, + testId, + aimockPort, + }) => { + // A's fixture streams slowly on purpose (see fixtures/queue/basic.json) + // so there's a window to send/cancel queued messages; give the whole + // scenario more room than the default 30s test timeout. + test.setTimeout(60_000) + + await page.goto(featureUrl('openai', 'chat', testId, aimockPort)) + + // Send A and wait for its stream to start. + await sendMessage(page, '[queue] first message') + await page + .getByTestId('loading-indicator') + .waitFor({ state: 'visible', timeout: 10_000 }) + + // While A is still streaming, send B then C. Both should be queued + // rather than starting a second concurrent request. + await sendMessage(page, '[queue] second message') + await sendMessage(page, '[queue] third message') + + const queued = page.getByTestId('queued-message') + await expect(queued).toHaveCount(2) + await expect(queued.nth(0)).toContainText('second message') + await expect(queued.nth(1)).toContainText('third message') + + // The queue region is distinct from the delivered message list — only + // A's user message has landed there so far. + await expect(page.getByTestId('user-message')).toHaveCount(1) + + // Cancel B before it drains. + await queued + .filter({ hasText: 'second message' }) + .getByTestId('cancel-queued-button') + .click() + + await expect(queued).toHaveCount(1) + await expect(queued.first()).toContainText('third message') + + // Let A's stream settle. The queue only auto-drains C once A's run fully + // completes, and draining adds C's user message immediately (before its + // own response streams back) — so waiting for a second user message is + // a reliable settle signal that doesn't fire early on A's own streamed + // text (which contains the substring "first response" from its very + // first chunk, long before the run actually finishes). + await expect(page.getByTestId('user-message')).toHaveCount(2, { + timeout: 40_000, + }) + await waitForAssistantText(page, 'first response', 5_000) + await expect(queued).toHaveCount(0) + await waitForAssistantText(page, 'third response', 20_000) + + // Final conversation: exactly A and C, in order — B was never sent. + const userMessages = page.getByTestId('user-message') + await expect(userMessages).toHaveCount(2) + await expect(userMessages.nth(0)).toContainText('first message') + await expect(userMessages.nth(1)).toContainText('third message') + await expect( + page.getByTestId('user-message').filter({ hasText: 'second message' }), + ).toHaveCount(0) + + const assistantMessages = page.getByTestId('assistant-message') + await expect(assistantMessages).toHaveCount(2) + await expect(assistantMessages.nth(0)).toContainText('first response') + await expect(assistantMessages.nth(1)).toContainText('third response') + + // Cross-check ordering across the whole transcript, not just per-role + // counts — proves A's turn fully precedes C's turn in the DOM. + const transcript = await page.getByTestId('message-list').innerText() + const iFirstMsg = transcript.indexOf('first message') + const iFirstResp = transcript.indexOf('first response') + const iThirdMsg = transcript.indexOf('third message') + const iThirdResp = transcript.indexOf('third response') + expect(iFirstMsg).toBeGreaterThanOrEqual(0) + expect(iFirstResp).toBeGreaterThan(iFirstMsg) + expect(iThirdMsg).toBeGreaterThan(iFirstResp) + expect(iThirdResp).toBeGreaterThan(iThirdMsg) + expect(transcript).not.toContain('second message') + }) +}) From c1651c9d8d5313daf38ca46d5808d00147104c8e Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 16:20:01 +0200 Subject: [PATCH 17/23] feat(ai-preact): expose queue + cancelQueued + per-send whenBusy Mirrors the ai-react implementation (17bb417d): ChatClient's queue-by-default behavior is now wired through the Preact useChat hook via onQueueChange, with cancelQueued and per-send whenBusy passthrough. Updates the two tests that previously asserted a mid-stream second sendMessage was dropped to instead assert both messages land in order. --- packages/ai-preact/src/types.ts | 26 +++++++++++++++++++++-- packages/ai-preact/src/use-chat.ts | 26 +++++++++++++++++++++-- packages/ai-preact/tests/use-chat.test.ts | 21 ++++++++++++------ 3 files changed, 63 insertions(+), 10 deletions(-) diff --git a/packages/ai-preact/src/types.ts b/packages/ai-preact/src/types.ts index faa04edae..73060a44f 100644 --- a/packages/ai-preact/src/types.ts +++ b/packages/ai-preact/src/types.ts @@ -9,11 +9,19 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueuedMessage, UIMessage, + WhenBusy, } from '@tanstack/ai-client' // Re-export types from ai-client -export type { ChatRequestBody, MultimodalContent, UIMessage } +export type { + ChatRequestBody, + MultimodalContent, + QueuedMessage, + UIMessage, + WhenBusy, +} /** * Options for the useChat hook. @@ -43,6 +51,7 @@ export type UseChatOptions< | 'onSubscriptionChange' | 'onConnectionStatusChange' | 'onSessionGeneratingChange' + | 'onQueueChange' | 'context' | 'devtools' > & { @@ -70,7 +79,20 @@ export interface UseChatReturn< * Send a message and get a response. * Can be a simple string or multimodal content with images, audio, etc. */ - sendMessage: (content: string | MultimodalContent) => Promise + sendMessage: ( + content: string | MultimodalContent, + options?: { whenBusy?: WhenBusy }, + ) => Promise + + /** + * Pending messages queued while a stream is in flight. + */ + queue: Array + + /** + * Cancel a queued message before it drains. No-op if already sent. + */ + cancelQueued: (id: string) => void /** * Append a message to the conversation diff --git a/packages/ai-preact/src/use-chat.ts b/packages/ai-preact/src/use-chat.ts index e10cb171d..b5038a53a 100644 --- a/packages/ai-preact/src/use-chat.ts +++ b/packages/ai-preact/src/use-chat.ts @@ -12,6 +12,8 @@ import type { ChatClientState, ConnectionStatus, InferredClientContext, + QueuedMessage, + WhenBusy, } from '@tanstack/ai-client' import type { AnyClientTool, ModelMessage } from '@tanstack/ai' @@ -39,6 +41,7 @@ export function useChat< const [connectionStatus, setConnectionStatus] = useState('disconnected') const [sessionGenerating, setSessionGenerating] = useState(false) + const [queue, setQueue] = useState>([]) // Track current messages in a ref to preserve them when client is recreated const messagesRef = useRef>>( @@ -151,6 +154,13 @@ export function useChat< if (activeClientRef.current !== instance) return setSessionGenerating(isGenerating) }, + ...(optionsRef.current.queue !== undefined && { + queue: optionsRef.current.queue, + }), + onQueueChange: (nextQueue: Array) => { + if (activeClientRef.current !== instance) return + setQueue(nextQueue) + }, }) activeClientRef.current = instance return instance @@ -221,8 +231,18 @@ export function useChat< // All callback options are read through optionsRef at call time, so fresh // closures from each render are picked up without recreating the client. const sendMessage = useCallback( - async (content: string | MultimodalContent) => { - await client.sendMessage(content) + async ( + content: string | MultimodalContent, + sendOptions?: { whenBusy?: WhenBusy }, + ) => { + await client.sendMessage(content, undefined, sendOptions) + }, + [client], + ) + + const cancelQueued = useCallback( + (id: string) => { + client.cancelQueued(id) }, [client], ) @@ -291,5 +311,7 @@ export function useChat< clear, addToolResult, addToolApprovalResponse, + queue, + cancelQueued, } } diff --git a/packages/ai-preact/tests/use-chat.test.ts b/packages/ai-preact/tests/use-chat.test.ts index 11e874c41..758cc2a60 100644 --- a/packages/ai-preact/tests/use-chat.test.ts +++ b/packages/ai-preact/tests/use-chat.test.ts @@ -397,7 +397,7 @@ describe('useChat', () => { expect(result.current.messages.length).toBe(0) }) - it('should not send message while loading', async () => { + it('should queue a message sent while loading and send it after', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 100, @@ -410,11 +410,16 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) }) - // Should only have one user message since second was blocked + // The second send is queued (default `whenBusy: 'queue'`) while the + // first stream is in flight, then auto-drains once it settles — both + // end up sent, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle errors during sendMessage', async () => { @@ -1261,7 +1266,7 @@ describe('useChat', () => { }) describe('concurrent operations', () => { - it('should handle multiple sendMessage calls', async () => { + it('should queue and then deliver multiple sendMessage calls in order', async () => { const adapter = createMockConnectionAdapter({ chunks: createTextChunks('Response'), chunkDelay: 50, @@ -1274,11 +1279,15 @@ describe('useChat', () => { await Promise.all([promise1, promise2]) }) - // Should only have one user message (second should be blocked) + // The second call is queued while the first stream is in flight, + // then auto-sent once it settles — both land, in order. const userMessages = result.current.messages.filter( (m) => m.role === 'user', ) - expect(userMessages.length).toBe(1) + expect(userMessages.map((m) => m.parts[0])).toEqual([ + { type: 'text', content: 'First' }, + { type: 'text', content: 'Second' }, + ]) }) it('should handle stop during sendMessage', async () => { From deefebe85d84727bfe5cdfc0cee752a9e092624f Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 16:23:03 +0200 Subject: [PATCH 18/23] feat: re-export queue config types from framework packages + add ai-preact to changeset --- .changeset/message-queue.md | 1 + packages/ai-preact/src/types.ts | 6 ++++++ packages/ai-react/src/types.ts | 6 ++++++ packages/ai-solid/src/types.ts | 6 ++++++ packages/ai-svelte/src/types.ts | 6 ++++++ packages/ai-vue/src/types.ts | 6 ++++++ 6 files changed, 31 insertions(+) diff --git a/.changeset/message-queue.md b/.changeset/message-queue.md index d2f440907..6f91ebae0 100644 --- a/.changeset/message-queue.md +++ b/.changeset/message-queue.md @@ -4,6 +4,7 @@ '@tanstack/ai-solid': minor '@tanstack/ai-vue': minor '@tanstack/ai-svelte': minor +'@tanstack/ai-preact': minor --- Messages sent while a stream is already in flight are now queued by default and automatically sent once the in-flight stream settles, instead of being silently dropped. **This is a behavior change.** diff --git a/packages/ai-preact/src/types.ts b/packages/ai-preact/src/types.ts index 73060a44f..b7334678a 100644 --- a/packages/ai-preact/src/types.ts +++ b/packages/ai-preact/src/types.ts @@ -9,7 +9,10 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } from '@tanstack/ai-client' @@ -18,7 +21,10 @@ import type { export type { ChatRequestBody, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } diff --git a/packages/ai-react/src/types.ts b/packages/ai-react/src/types.ts index 429b08262..fdd7b7958 100644 --- a/packages/ai-react/src/types.ts +++ b/packages/ai-react/src/types.ts @@ -14,7 +14,10 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } from '@tanstack/ai-client' @@ -23,7 +26,10 @@ import type { export type { ChatRequestBody, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } diff --git a/packages/ai-solid/src/types.ts b/packages/ai-solid/src/types.ts index c3466a096..1aa30084d 100644 --- a/packages/ai-solid/src/types.ts +++ b/packages/ai-solid/src/types.ts @@ -14,7 +14,10 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } from '@tanstack/ai-client' @@ -24,7 +27,10 @@ import type { Accessor } from 'solid-js' export type { ChatRequestBody, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } diff --git a/packages/ai-svelte/src/types.ts b/packages/ai-svelte/src/types.ts index f7bf3ae83..7ec653a13 100644 --- a/packages/ai-svelte/src/types.ts +++ b/packages/ai-svelte/src/types.ts @@ -14,7 +14,10 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } from '@tanstack/ai-client' @@ -23,7 +26,10 @@ import type { export type { ChatRequestBody, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } diff --git a/packages/ai-vue/src/types.ts b/packages/ai-vue/src/types.ts index 0235df297..2fa75b8d1 100644 --- a/packages/ai-vue/src/types.ts +++ b/packages/ai-vue/src/types.ts @@ -14,7 +14,10 @@ import type { DistributedOmit, InferredClientContext, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } from '@tanstack/ai-client' @@ -24,7 +27,10 @@ import type { DeepReadonly, ShallowRef } from 'vue' export type { ChatRequestBody, MultimodalContent, + QueueConfig, QueuedMessage, + QueueOption, + QueueStrategy, UIMessage, WhenBusy, } From ddd52fc726caf0eee6ef7932fa6955b6d8275ae7 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 16:23:41 +0200 Subject: [PATCH 19/23] test(ai-client): cover multimodal batch-drain merge; docs: clarify whenBusy shorthand --- docs/chat/streaming.md | 2 +- .../ai-client/tests/chat-client-queue.test.ts | 47 +++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/docs/chat/streaming.md b/docs/chat/streaming.md index 8382629bb..a4159ac50 100644 --- a/docs/chat/streaming.md +++ b/docs/chat/streaming.md @@ -224,7 +224,7 @@ const { messages, queue, sendMessage, cancelQueued, isLoading } = useChat({ - **`maxSize`** — caps how many messages can be queued. - **`onOverflow`** — `"reject"` (default) ignores a send once `maxSize` is reached; `"drop-oldest"` evicts the oldest queued item to make room. -You can also pass a plain `WhenBusy` string as shorthand for `queue: "interrupt"`, or a `QueueStrategy` function for full control over the per-send decision (the drain order stays FIFO for the function form). +You can also pass a plain `WhenBusy` string (e.g. `queue: "interrupt"`) as shorthand for `{ whenBusy: "interrupt" }` — this applies for any `WhenBusy` value (`"queue"`, `"drop"`, or `"interrupt"`), not just `"interrupt"` — or a `QueueStrategy` function for full control over the per-send decision (the drain order stays FIFO for the function form). `useChat` exposes the pending queue as `queue` so you can render it distinctly from `messages`, along with `cancelQueued(id)` to cancel an item before it sends: diff --git a/packages/ai-client/tests/chat-client-queue.test.ts b/packages/ai-client/tests/chat-client-queue.test.ts index ab3dfbf39..3bbb48288 100644 --- a/packages/ai-client/tests/chat-client-queue.test.ts +++ b/packages/ai-client/tests/chat-client-queue.test.ts @@ -246,6 +246,53 @@ describe('ChatClient queue drain', () => { }) }) + it('batch drain flattens multimodal queued items into ContentPart[]', async () => { + const { connection, release } = createSequencedHoldingConnection() + const client = new ChatClient({ connection, queue: { drain: 'batch' } }) + + const firstSend = client.sendMessage('first') + await vi.waitFor(() => { + expect(client.getIsLoading()).toBe(true) + }) + + // Queued item 1: multimodal content (array of ContentPart). + await client.sendMessage({ + content: [ + { type: 'text', content: 'look' }, + { + type: 'image', + source: { type: 'url', value: 'https://example.com/a.png' }, + }, + ], + }) + // Queued item 2: plain string content. + await client.sendMessage('note') + expect(client.getQueue()).toHaveLength(2) + + release() + await firstSend + + expect(client.getQueue()).toEqual([]) + const userMessages = client + .getMessages() + .filter((m) => m.role === 'user') + // 'first' streamed immediately; the two queued sends above are merged + // into a single batched send, so there should be exactly 2 user messages. + expect(userMessages).toHaveLength(2) + // The merged message's parts must be the multimodal item's parts + // (flattened, in order) followed by the string item flattened to text — + // proving the MULTIMODAL branch of `mergeQueuedMessages` ran, not just + // the all-string join. + expect(userMessages[1]?.parts).toEqual([ + { type: 'text', content: 'look' }, + { + type: 'image', + source: { type: 'url', value: 'https://example.com/a.png' }, + }, + { type: 'text', content: 'note' }, + ]) + }) + it('flushes the queue on stop()', async () => { const { connection, release } = createHoldingConnection() const client = new ChatClient({ connection }) From c551ae028fe21c0b7ea465bdbac922b3594e0583 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 16:36:49 +0200 Subject: [PATCH 20/23] fix: sort queue-type imports alphabetically in framework types --- packages/ai-preact/src/types.ts | 2 +- packages/ai-react/src/types.ts | 2 +- packages/ai-solid/src/types.ts | 2 +- packages/ai-svelte/src/types.ts | 2 +- packages/ai-vue/src/types.ts | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/ai-preact/src/types.ts b/packages/ai-preact/src/types.ts index b7334678a..5adb94dee 100644 --- a/packages/ai-preact/src/types.ts +++ b/packages/ai-preact/src/types.ts @@ -10,9 +10,9 @@ import type { InferredClientContext, MultimodalContent, QueueConfig, - QueuedMessage, QueueOption, QueueStrategy, + QueuedMessage, UIMessage, WhenBusy, } from '@tanstack/ai-client' diff --git a/packages/ai-react/src/types.ts b/packages/ai-react/src/types.ts index fdd7b7958..7a429ec13 100644 --- a/packages/ai-react/src/types.ts +++ b/packages/ai-react/src/types.ts @@ -15,9 +15,9 @@ import type { InferredClientContext, MultimodalContent, QueueConfig, - QueuedMessage, QueueOption, QueueStrategy, + QueuedMessage, UIMessage, WhenBusy, } from '@tanstack/ai-client' diff --git a/packages/ai-solid/src/types.ts b/packages/ai-solid/src/types.ts index 1aa30084d..80c8aa7aa 100644 --- a/packages/ai-solid/src/types.ts +++ b/packages/ai-solid/src/types.ts @@ -15,9 +15,9 @@ import type { InferredClientContext, MultimodalContent, QueueConfig, - QueuedMessage, QueueOption, QueueStrategy, + QueuedMessage, UIMessage, WhenBusy, } from '@tanstack/ai-client' diff --git a/packages/ai-svelte/src/types.ts b/packages/ai-svelte/src/types.ts index 7ec653a13..beb745fad 100644 --- a/packages/ai-svelte/src/types.ts +++ b/packages/ai-svelte/src/types.ts @@ -15,9 +15,9 @@ import type { InferredClientContext, MultimodalContent, QueueConfig, - QueuedMessage, QueueOption, QueueStrategy, + QueuedMessage, UIMessage, WhenBusy, } from '@tanstack/ai-client' diff --git a/packages/ai-vue/src/types.ts b/packages/ai-vue/src/types.ts index 2fa75b8d1..80ce3d6a6 100644 --- a/packages/ai-vue/src/types.ts +++ b/packages/ai-vue/src/types.ts @@ -15,9 +15,9 @@ import type { InferredClientContext, MultimodalContent, QueueConfig, - QueuedMessage, QueueOption, QueueStrategy, + QueuedMessage, UIMessage, WhenBusy, } from '@tanstack/ai-client' From 6aa7207dc1085705511ef54a2933cdb1ce51dc16 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:51:45 +0000 Subject: [PATCH 21/23] ci: apply automated fixes --- .../ai-client/tests/chat-client-queue.test.ts | 25 +++++-------------- 1 file changed, 6 insertions(+), 19 deletions(-) diff --git a/packages/ai-client/tests/chat-client-queue.test.ts b/packages/ai-client/tests/chat-client-queue.test.ts index 3bbb48288..1137ac299 100644 --- a/packages/ai-client/tests/chat-client-queue.test.ts +++ b/packages/ai-client/tests/chat-client-queue.test.ts @@ -131,9 +131,7 @@ describe('ChatClient message queue', () => { // The dropped message must never have become a real message either — // it should be gone entirely, not just missing from the queue. - const userMessages = client - .getMessages() - .filter((m) => m.role === 'user') + const userMessages = client.getMessages().filter((m) => m.role === 'user') expect(userMessages.map((m) => m.parts[0])).toEqual([ { type: 'text', content: 'first' }, ]) @@ -200,18 +198,13 @@ describe('ChatClient queue drain', () => { await client.sendMessage('second') await client.sendMessage('third') - expect(client.getQueue().map((m) => m.content)).toEqual([ - 'second', - 'third', - ]) + expect(client.getQueue().map((m) => m.content)).toEqual(['second', 'third']) release() await firstSend expect(client.getQueue()).toEqual([]) - const userMessages = client - .getMessages() - .filter((m) => m.role === 'user') + const userMessages = client.getMessages().filter((m) => m.role === 'user') expect(userMessages.map((m) => m.parts[0])).toEqual([ { type: 'text', content: 'first' }, { type: 'text', content: 'second' }, @@ -236,9 +229,7 @@ describe('ChatClient queue drain', () => { await firstSend expect(client.getQueue()).toEqual([]) - const userMessages = client - .getMessages() - .filter((m) => m.role === 'user') + const userMessages = client.getMessages().filter((m) => m.role === 'user') expect(userMessages).toHaveLength(2) expect(userMessages[1]?.parts[0]).toEqual({ type: 'text', @@ -273,9 +264,7 @@ describe('ChatClient queue drain', () => { await firstSend expect(client.getQueue()).toEqual([]) - const userMessages = client - .getMessages() - .filter((m) => m.role === 'user') + const userMessages = client.getMessages().filter((m) => m.role === 'user') // 'first' streamed immediately; the two queued sends above are merged // into a single batched send, so there should be exactly 2 user messages. expect(userMessages).toHaveLength(2) @@ -331,9 +320,7 @@ describe('ChatClient queue drain', () => { // The queue must be flushed, not stranded or auto-drained into the // now-broken endpoint. expect(client.getQueue()).toEqual([]) - const userMessages = client - .getMessages() - .filter((m) => m.role === 'user') + const userMessages = client.getMessages().filter((m) => m.role === 'user') expect(userMessages.map((m) => m.parts[0])).toEqual([ { type: 'text', content: 'first' }, ]) From cb34848e0fae9060c5524d05b1c0d014620ef116 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 17:12:23 +0200 Subject: [PATCH 22/23] fix: export queue types from framework package entrypoints The framework packages re-exported QueuedMessage/WhenBusy/QueueConfig/ QueueStrategy/QueueOption from their internal types module but never forwarded them through the public index entry, so importing them from e.g. @tanstack/ai-react failed at the package boundary. --- packages/ai-preact/src/index.ts | 5 +++++ packages/ai-react/src/index.ts | 5 +++++ packages/ai-solid/src/index.ts | 5 +++++ packages/ai-svelte/src/index.ts | 5 +++++ packages/ai-vue/src/index.ts | 5 +++++ 5 files changed, 25 insertions(+) diff --git a/packages/ai-preact/src/index.ts b/packages/ai-preact/src/index.ts index 9362a2650..a5175b58a 100644 --- a/packages/ai-preact/src/index.ts +++ b/packages/ai-preact/src/index.ts @@ -6,6 +6,11 @@ export type { UseChatReturn, UIMessage, ChatRequestBody, + QueuedMessage, + WhenBusy, + QueueConfig, + QueueStrategy, + QueueOption, } from './types' export { diff --git a/packages/ai-react/src/index.ts b/packages/ai-react/src/index.ts index f8842333a..e74cefa2d 100644 --- a/packages/ai-react/src/index.ts +++ b/packages/ai-react/src/index.ts @@ -8,6 +8,11 @@ export type { UseChatReturn, UIMessage, ChatRequestBody, + QueuedMessage, + WhenBusy, + QueueConfig, + QueueStrategy, + QueueOption, } from './types' export type { UseRealtimeChatOptions, diff --git a/packages/ai-solid/src/index.ts b/packages/ai-solid/src/index.ts index f94e214f1..bc18d6187 100644 --- a/packages/ai-solid/src/index.ts +++ b/packages/ai-solid/src/index.ts @@ -5,6 +5,11 @@ export type { UseChatReturn, UIMessage, ChatRequestBody, + QueuedMessage, + WhenBusy, + QueueConfig, + QueueStrategy, + QueueOption, } from './types' // Generation hooks diff --git a/packages/ai-svelte/src/index.ts b/packages/ai-svelte/src/index.ts index d2dd25cbd..e2a1a6a39 100644 --- a/packages/ai-svelte/src/index.ts +++ b/packages/ai-svelte/src/index.ts @@ -5,6 +5,11 @@ export type { DeepPartial, UIMessage, ChatRequestBody, + QueuedMessage, + WhenBusy, + QueueConfig, + QueueStrategy, + QueueOption, } from './types' // Generation hooks diff --git a/packages/ai-vue/src/index.ts b/packages/ai-vue/src/index.ts index f94e214f1..bc18d6187 100644 --- a/packages/ai-vue/src/index.ts +++ b/packages/ai-vue/src/index.ts @@ -5,6 +5,11 @@ export type { UseChatReturn, UIMessage, ChatRequestBody, + QueuedMessage, + WhenBusy, + QueueConfig, + QueueStrategy, + QueueOption, } from './types' // Generation hooks From 7401f077d59fa4d51db015260b87209ff3f956e6 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 6 Jul 2026 17:12:23 +0200 Subject: [PATCH 23/23] docs(example): add queueing-strategies route to ts-react-chat Side-by-side panels (queue/fifo, interrupt, queue/batch) driven by a shared composer, showcasing how each queue strategy handles a message sent while a stream is in flight, with live queue rendering + cancel. --- .../ts-react-chat/src/components/Header.tsx | 14 + examples/ts-react-chat/src/routeTree.gen.ts | 21 ++ .../ts-react-chat/src/routes/queueing.tsx | 288 ++++++++++++++++++ 3 files changed, 323 insertions(+) create mode 100644 examples/ts-react-chat/src/routes/queueing.tsx diff --git a/examples/ts-react-chat/src/components/Header.tsx b/examples/ts-react-chat/src/components/Header.tsx index e7fbc6a0a..090a8d9b4 100644 --- a/examples/ts-react-chat/src/components/Header.tsx +++ b/examples/ts-react-chat/src/components/Header.tsx @@ -10,6 +10,7 @@ import { Guitar, Home, Image, + Layers, Menu, LayoutGrid, Mic, @@ -231,6 +232,19 @@ export default function Header() { Persistent Chats + setIsOpen(false)} + className="flex items-center gap-3 p-3 rounded-lg hover:bg-gray-800 transition-colors mb-2" + activeProps={{ + className: + 'flex items-center gap-3 p-3 rounded-lg bg-cyan-600 hover:bg-cyan-700 transition-colors mb-2', + }} + > + + Queueing Strategies + + setIsOpen(false)} diff --git a/examples/ts-react-chat/src/routeTree.gen.ts b/examples/ts-react-chat/src/routeTree.gen.ts index 69a7764a2..a66c65d4c 100644 --- a/examples/ts-react-chat/src/routeTree.gen.ts +++ b/examples/ts-react-chat/src/routeTree.gen.ts @@ -13,6 +13,7 @@ import { Route as ThreadsRouteImport } from './routes/threads' import { Route as ServerFnChatRouteImport } from './routes/server-fn-chat' import { Route as SandboxesRouteImport } from './routes/sandboxes' import { Route as RealtimeRouteImport } from './routes/realtime' +import { Route as QueueingRouteImport } from './routes/queueing' import { Route as McpDemoRouteImport } from './routes/mcp-demo' import { Route as McpAppsRouteImport } from './routes/mcp-apps' import { Route as Issue176ToolResultRouteImport } from './routes/issue-176-tool-result' @@ -74,6 +75,11 @@ const RealtimeRoute = RealtimeRouteImport.update({ path: '/realtime', getParentRoute: () => rootRouteImport, } as any) +const QueueingRoute = QueueingRouteImport.update({ + id: '/queueing', + path: '/queueing', + getParentRoute: () => rootRouteImport, +} as any) const McpDemoRoute = McpDemoRouteImport.update({ id: '/mcp-demo', path: '/mcp-demo', @@ -287,6 +293,7 @@ export interface FileRoutesByFullPath { '/issue-176-tool-result': typeof Issue176ToolResultRoute '/mcp-apps': typeof McpAppsRoute '/mcp-demo': typeof McpDemoRoute + '/queueing': typeof QueueingRoute '/realtime': typeof RealtimeRoute '/sandboxes': typeof SandboxesRoute '/server-fn-chat': typeof ServerFnChatRoute @@ -333,6 +340,7 @@ export interface FileRoutesByTo { '/issue-176-tool-result': typeof Issue176ToolResultRoute '/mcp-apps': typeof McpAppsRoute '/mcp-demo': typeof McpDemoRoute + '/queueing': typeof QueueingRoute '/realtime': typeof RealtimeRoute '/sandboxes': typeof SandboxesRoute '/server-fn-chat': typeof ServerFnChatRoute @@ -380,6 +388,7 @@ export interface FileRoutesById { '/issue-176-tool-result': typeof Issue176ToolResultRoute '/mcp-apps': typeof McpAppsRoute '/mcp-demo': typeof McpDemoRoute + '/queueing': typeof QueueingRoute '/realtime': typeof RealtimeRoute '/sandboxes': typeof SandboxesRoute '/server-fn-chat': typeof ServerFnChatRoute @@ -428,6 +437,7 @@ export interface FileRouteTypes { | '/issue-176-tool-result' | '/mcp-apps' | '/mcp-demo' + | '/queueing' | '/realtime' | '/sandboxes' | '/server-fn-chat' @@ -474,6 +484,7 @@ export interface FileRouteTypes { | '/issue-176-tool-result' | '/mcp-apps' | '/mcp-demo' + | '/queueing' | '/realtime' | '/sandboxes' | '/server-fn-chat' @@ -520,6 +531,7 @@ export interface FileRouteTypes { | '/issue-176-tool-result' | '/mcp-apps' | '/mcp-demo' + | '/queueing' | '/realtime' | '/sandboxes' | '/server-fn-chat' @@ -567,6 +579,7 @@ export interface RootRouteChildren { Issue176ToolResultRoute: typeof Issue176ToolResultRoute McpAppsRoute: typeof McpAppsRoute McpDemoRoute: typeof McpDemoRoute + QueueingRoute: typeof QueueingRoute RealtimeRoute: typeof RealtimeRoute SandboxesRoute: typeof SandboxesRoute ServerFnChatRoute: typeof ServerFnChatRoute @@ -635,6 +648,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof RealtimeRouteImport parentRoute: typeof rootRouteImport } + '/queueing': { + id: '/queueing' + path: '/queueing' + fullPath: '/queueing' + preLoaderRoute: typeof QueueingRouteImport + parentRoute: typeof rootRouteImport + } '/mcp-demo': { id: '/mcp-demo' path: '/mcp-demo' @@ -927,6 +947,7 @@ const rootRouteChildren: RootRouteChildren = { Issue176ToolResultRoute: Issue176ToolResultRoute, McpAppsRoute: McpAppsRoute, McpDemoRoute: McpDemoRoute, + QueueingRoute: QueueingRoute, RealtimeRoute: RealtimeRoute, SandboxesRoute: SandboxesRoute, ServerFnChatRoute: ServerFnChatRoute, diff --git a/examples/ts-react-chat/src/routes/queueing.tsx b/examples/ts-react-chat/src/routes/queueing.tsx new file mode 100644 index 000000000..de6a340ba --- /dev/null +++ b/examples/ts-react-chat/src/routes/queueing.tsx @@ -0,0 +1,288 @@ +import { useMemo, useState } from 'react' +import { createFileRoute } from '@tanstack/react-router' +import { Send, Square, X, Zap } from 'lucide-react' +import { fetchServerSentEvents, useChat } from '@tanstack/ai-react' +import type { QueueConfig, QueuedMessage, UIMessage } from '@tanstack/ai-react' +import { DEFAULT_MODEL_OPTION, MODEL_OPTIONS } from '@/lib/model-selection' +import type { ModelOption } from '@/lib/model-selection' + +/** + * Showcase route for the client-side message queue. + * + * Three chats run side by side against the same endpoint, each hardwired to a + * different `queue` strategy. A shared composer broadcasts the same message to + * all three at once, so you can send a message, then send another WHILE the + * first is still streaming, and watch each strategy diverge: + * + * - queue/fifo → the second message is held and auto-sent after the first + * settles; it shows up as a cancellable pending item. + * - interrupt → the in-flight stream is aborted and the new message sends + * immediately (the queue never fills). + * - queue/batch → multiple mid-stream messages are held, then merged into a + * single send when the stream settles. + */ + +interface StrategyMeta { + title: string + blurb: string + config: QueueConfig +} + +const FIFO: StrategyMeta = { + title: 'queue · fifo', + blurb: + 'Messages sent while streaming are held and auto-sent one at a time, in order. Pending messages are cancellable until they drain.', + config: { whenBusy: 'queue', drain: 'fifo' }, +} +const INTERRUPT: StrategyMeta = { + title: 'interrupt', + blurb: + 'A message sent while streaming aborts the in-flight response and sends immediately. The queue never fills.', + config: { whenBusy: 'interrupt' }, +} +const BATCH: StrategyMeta = { + title: 'queue · batch', + blurb: + 'Messages sent while streaming are held, then merged into a single send (joined by newlines) when the stream settles.', + config: { whenBusy: 'queue', drain: 'batch' }, +} + +function textOf(message: UIMessage): string { + return message.parts + .filter((part) => part.type === 'text') + .map((part) => (part.type === 'text' ? part.content : '')) + .join('') +} + +interface PanelChat { + messages: Array + isLoading: boolean + queue: Array + cancelQueued: (id: string) => void + stop: () => void + error: Error | undefined +} + +function StrategyPanel({ + strategy, + chat, +}: { + strategy: StrategyMeta + chat: PanelChat +}) { + const { messages, isLoading, queue, cancelQueued, stop, error } = chat + const visible = messages.filter( + (message) => textOf(message).trim().length > 0, + ) + + return ( +
+
+
+ + {strategy.title} + + {isLoading && ( + + )} +
+

+ {strategy.blurb} +

+
+ +
+ {visible.length === 0 && ( +

No messages yet.

+ )} + {visible.map((message) => ( +
+ + {message.role === 'assistant' ? 'AI' : 'You'}: + + {textOf(message)} +
+ ))} + {isLoading && ( +

⏳ streaming…

+ )} +
+ + {error && ( +
+ {error.message} +
+ )} + +
+

+ Queue ({queue.length}) +

+ {queue.length === 0 ? ( +

empty

+ ) : ( +
    + {queue.map((item) => ( +
  • + + {typeof item.content === 'string' + ? item.content + : '[multimodal]'} + + +
  • + ))} +
+ )} +
+
+ ) +} + +function QueueingPage() { + const [selectedModel, setSelectedModel] = + useState(DEFAULT_MODEL_OPTION) + const [input, setInput] = useState('') + + const body = useMemo( + () => ({ provider: selectedModel.provider, model: selectedModel.model }), + [selectedModel.provider, selectedModel.model], + ) + + // One chat per strategy, called explicitly (fixed count, fixed order) so the + // rules of hooks hold. The shared composer broadcasts the same text to all + // three via their sendMessage functions. + const fifo = useChat({ + connection: fetchServerSentEvents('/api/tanchat'), + body, + queue: FIFO.config, + }) + const interrupt = useChat({ + connection: fetchServerSentEvents('/api/tanchat'), + body, + queue: INTERRUPT.config, + }) + const batch = useChat({ + connection: fetchServerSentEvents('/api/tanchat'), + body, + queue: BATCH.config, + }) + + const panels = [ + { meta: FIFO, chat: fifo }, + { meta: INTERRUPT, chat: interrupt }, + { meta: BATCH, chat: batch }, + ] + + const broadcast = () => { + const text = input.trim() + if (!text) return + void fifo.sendMessage(text) + void interrupt.sendMessage(text) + void batch.sendMessage(text) + setInput('') + } + + return ( +
+
+

+ Queueing Strategies +

+

+ Send a message, then send another while it is still streaming{' '} + — each panel reacts differently based on its queue{' '} + config. +

+
+ + +
+
+ +
+ {panels.map((panel) => ( + + ))} +
+ +
+
+