From c74aa74c1def9adc976ed8f3c0f39b71019db907 Mon Sep 17 00:00:00 2001 From: kolaworld Date: Sun, 16 Aug 2026 11:57:36 -0400 Subject: [PATCH 1/6] fix(ai): identify hydrated interrupt state changes Propagate hydrate and live context through interrupt state callbacks, including synchronous React and Preact restoration. Fixes #1086 --- .../fix-1086-interrupt-hydrate-source.md | 11 +++ packages/ai-angular/src/inject-chat.ts | 4 +- packages/ai-angular/tests/inject-chat.test.ts | 2 + packages/ai-client/src/chat-client.ts | 63 +++++++----- packages/ai-client/src/interrupt-manager.ts | 26 +++-- packages/ai-client/src/types.ts | 11 ++- .../tests/chat-client-interrupts.test.ts | 46 ++++++++- .../tests/interrupts-types.test-d.ts | 3 +- .../ai-client/tests/resume-snapshot.test.ts | 98 ++++++++++++++++++- packages/ai-preact/src/use-chat.ts | 50 ++++++---- packages/ai-preact/tests/use-chat.test.ts | 32 ++++++ packages/ai-react/src/use-chat.ts | 50 ++++++---- packages/ai-react/tests/use-chat.test.ts | 50 +++++++++- packages/ai-solid/src/use-chat.ts | 4 +- packages/ai-solid/tests/use-chat.test.ts | 1 + packages/ai-svelte/src/create-chat.svelte.ts | 4 +- packages/ai-svelte/tests/use-chat.test.ts | 1 + packages/ai-vue/src/use-chat.ts | 4 +- packages/ai-vue/tests/use-chat.test.ts | 1 + .../e2e/src/routes/persistence-durability.tsx | 9 ++ .../e2e/tests/persistence-durability.spec.ts | 3 + 21 files changed, 387 insertions(+), 86 deletions(-) create mode 100644 .changeset/fix-1086-interrupt-hydrate-source.md diff --git a/.changeset/fix-1086-interrupt-hydrate-source.md b/.changeset/fix-1086-interrupt-hydrate-source.md new file mode 100644 index 0000000000..48a10895fc --- /dev/null +++ b/.changeset/fix-1086-interrupt-hydrate-source.md @@ -0,0 +1,11 @@ +--- +'@tanstack/ai-client': patch +'@tanstack/ai-react': patch +'@tanstack/ai-solid': patch +'@tanstack/ai-vue': patch +'@tanstack/ai-svelte': patch +'@tanstack/ai-preact': patch +'@tanstack/ai-angular': patch +--- + +`onInterruptStateChange` now identifies snapshot restoration (`hydrate`) separately from streamed or client-initiated interrupt updates (`live`). The source follows each state publication, so cancelling a restored batch from the callback produces subsequent `live` updates without re-entering hydration. Client-tool interrupts remain hidden from the public list in both cases; `hydrate` lets an app cancel a restored batch without cancelling one that is still running. diff --git a/packages/ai-angular/src/inject-chat.ts b/packages/ai-angular/src/inject-chat.ts index 6e060038b7..59ba74435a 100644 --- a/packages/ai-angular/src/inject-chat.ts +++ b/packages/ai-angular/src/inject-chat.ts @@ -131,9 +131,9 @@ export function injectChat< // signal (via `onRunIdChange`) and pending interrupts arrive through // `onInterruptStateChange`, so there is nothing left for it to do — and it // is not a public option here, matching the other framework packages. - onInterruptStateChange: (nextInterruptState) => { + onInterruptStateChange: (nextInterruptState, context) => { interruptState.set(nextInterruptState) - options.onInterruptStateChange?.(nextInterruptState) + options.onInterruptStateChange?.(nextInterruptState, context) }, tools: options.tools, onCustomEvent: (eventType, data, context) => diff --git a/packages/ai-angular/tests/inject-chat.test.ts b/packages/ai-angular/tests/inject-chat.test.ts index 4586b1f9f2..520ab33557 100644 --- a/packages/ai-angular/tests/inject-chat.test.ts +++ b/packages/ai-angular/tests/inject-chat.test.ts @@ -59,6 +59,7 @@ describe('injectChat', () => { interrupts: result.interrupts(), interruptErrors: result.interruptErrors(), }), + { source: 'live' }, ) }) @@ -254,6 +255,7 @@ describe('injectChat — resume', () => { expect.objectContaining({ id: 'interrupt-1' }), ]), }), + { source: 'live' }, ) }) }) diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index 5f1dcb06ff..3efafd4b26 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -59,7 +59,10 @@ import type { UIMessage, WhenBusy, } from './types' -import type { InterruptManagerSubmission } from './interrupt-manager' +import type { + InterruptManagerChangeSource, + InterruptManagerSubmission, +} from './interrupt-manager' /** Internal queue entry — public {@link QueuedMessage} plus optional per-send body. */ interface InternalQueuedMessage extends QueuedMessage { @@ -93,7 +96,10 @@ type ChatClientUpdateOptionsWithoutContext< * starts (including a rejoin), `null` when it settles. */ onRunIdChange?: (runId: string | null) => void - onInterruptStateChange?: (state: ChatInterruptState) => void + onInterruptStateChange?: ( + state: ChatInterruptState, + context: { source: 'hydrate' | 'live' }, + ) => void onCustomEvent?: ( eventType: string, data: unknown, @@ -400,7 +406,10 @@ export class ChatClient< pendingInterrupts: BoundInterrupts, ) => void onRunIdChange: (runId: string | null) => void - onInterruptStateChange: (state: ChatInterruptState) => void + onInterruptStateChange: ( + state: ChatInterruptState, + context: { source: 'hydrate' | 'live' }, + ) => void onCustomEvent: ( eventType: string, data: unknown, @@ -489,7 +498,7 @@ export class ChatClient< this.interruptManager = new InterruptManager({ ...(options.tools !== undefined ? { tools: options.tools } : {}), submit: (submission) => this.submitInterruptBatch(submission), - onChange: () => this.notifyResumeStateChange(), + onChange: (source) => this.notifyResumeStateChange(source), }) // In-memory rehydrate of interrupt descriptors (e.g. after a page reload @@ -848,7 +857,7 @@ export class ChatClient< private applyResumeSnapshot(snapshot: ChatResumeSnapshot): void { const resumeState = readResumeState(snapshot) if (resumeState === undefined) { - this.interruptManager.reset() + this.interruptManager.reset({ source: 'hydrate' }) return } this.lastResume = resumeState @@ -856,16 +865,19 @@ export class ChatClient< ? snapshot.pendingInterrupts : [] if (pendingInterrupts.length === 0) { - this.interruptManager.reset() + this.interruptManager.reset({ source: 'hydrate' }) return } const generation = this.interruptGeneration(pendingInterrupts) - this.interruptManager.hydrate({ - threadId: resumeState.threadId, - interruptedRunId: resumeState.runId, - generation, - interrupts: pendingInterrupts, - }) + this.interruptManager.hydrate( + { + threadId: resumeState.threadId, + interruptedRunId: resumeState.runId, + generation, + interrupts: pendingInterrupts, + }, + 'hydrate', + ) } /** @@ -1085,12 +1097,15 @@ export class ChatClient< threadId: threadId ?? this.threadId, runId: interruptedRunId, } - this.interruptManager.hydrate({ - threadId: this.lastResume.threadId, - interruptedRunId, - generation: this.interruptGeneration(chunk.outcome.interrupts), - interrupts: chunk.outcome.interrupts, - }) + this.interruptManager.hydrate( + { + threadId: this.lastResume.threadId, + interruptedRunId, + generation: this.interruptGeneration(chunk.outcome.interrupts), + interrupts: chunk.outcome.interrupts, + }, + 'live', + ) return } @@ -1137,7 +1152,7 @@ export class ChatClient< this.interruptManager.reset() return } - this.notifyResumeStateChange() + this.notifyResumeStateChange('live') } /** @@ -1345,18 +1360,22 @@ export class ChatClient< this.devtoolsBridge.emitSnapshot() } - private notifyResumeStateChange(): void { + private notifyResumeStateChange(source: InterruptManagerChangeSource): void { const resumeState = this.getResumeState() + // Capture state before invoking callbacks so a synchronous nested change + // cannot pair this publication's source with a later manager snapshot. + const interruptState = this.interruptManager.getState() // Persist (or clear) the durable resume snapshot so a full page reload can // rehydrate pending interrupts and rejoin the run. Folded into the same // persistence adapter that stores messages (one record per chat). this.persistResumeSnapshot(resumeState) this.callbacksRef.current.onResumeStateChange( resumeState, - this.interruptManager.getInterrupts(), + interruptState.interrupts, ) this.callbacksRef.current.onInterruptStateChange( - this.interruptManager.getState(), + interruptState, + { source }, ) } diff --git a/packages/ai-client/src/interrupt-manager.ts b/packages/ai-client/src/interrupt-manager.ts index 1524ad8bd3..79d6e01416 100644 --- a/packages/ai-client/src/interrupt-manager.ts +++ b/packages/ai-client/src/interrupt-manager.ts @@ -44,12 +44,14 @@ export interface InterruptManagerSubmission { fingerprint: string } +export type InterruptManagerChangeSource = 'hydrate' | 'live' + export interface InterruptManagerOptions< TTools extends ReadonlyArray, > { tools?: TTools submit: (submission: InterruptManagerSubmission) => Promise - onChange?: () => void + onChange?: (source: InterruptManagerChangeSource) => void } type UnknownObject = { [key: string]: unknown } @@ -515,7 +517,10 @@ export class InterruptManager< this.tools = tools } - hydrate(hydration: InterruptManagerHydration): void { + hydrate( + hydration: InterruptManagerHydration, + source: InterruptManagerChangeSource = 'live', + ): void { this.hydration = { threadId: hydration.threadId, interruptedRunId: hydration.interruptedRunId, @@ -529,7 +534,7 @@ export class InterruptManager< this.submissionRootErrors = Object.freeze([]) this.retrySubmission = undefined this.resuming = false - this.publish() + this.publish(source) } getInterrupts(): BoundInterrupts { @@ -544,7 +549,10 @@ export class InterruptManager< return this.hydration?.interrupts ?? Object.freeze([]) } - reset(options?: { preserveRootErrors?: boolean }): void { + reset(options?: { + preserveRootErrors?: boolean + source?: InterruptManagerChangeSource + }): void { this.hydration = undefined this.items = [] this.snapshot = Object.freeze([]) @@ -560,7 +568,7 @@ export class InterruptManager< interruptErrors: this.rootErrors, resuming: false, }) - this.options.onChange?.() + this.options.onChange?.(options?.source ?? 'live') } getInterruptErrors(): ReadonlyArray { @@ -854,7 +862,9 @@ export class InterruptManager< return Object.freeze(next) as BoundInterrupts } - private publish(): void { + // Provenance belongs to each publication because `onChange` may synchronously + // mutate the manager and publish again before an outer callback returns. + private publish(source: InterruptManagerChangeSource = 'live'): void { if (!this.hydration) { this.snapshot = Object.freeze([]) this.state = Object.freeze({ @@ -863,7 +873,7 @@ export class InterruptManager< interruptErrors: this.rootErrors, resuming: this.resuming, }) - this.options.onChange?.() + this.options.onChange?.(source) return } this.snapshot = this.buildSnapshot() @@ -873,7 +883,7 @@ export class InterruptManager< interruptErrors: this.rootErrors, resuming: this.resuming, }) - this.options.onChange?.() + this.options.onChange?.(source) } private resolveItem( diff --git a/packages/ai-client/src/types.ts b/packages/ai-client/src/types.ts index 75ff059763..c8003ae10d 100644 --- a/packages/ai-client/src/types.ts +++ b/packages/ai-client/src/types.ts @@ -880,8 +880,15 @@ export interface ChatClientBaseOptions< */ onRunIdChange?: (runId: string | null) => void - /** Callback when the immutable interrupt state snapshot changes. */ - onInterruptStateChange?: (state: ChatInterruptState) => void + /** + * Callback when the immutable interrupt state snapshot changes. + * Snapshot restoration passes `{ source: 'hydrate' }`; streamed and + * client-initiated updates pass `{ source: 'live' }`. + */ + onInterruptStateChange?: ( + state: ChatInterruptState, + context: { source: 'hydrate' | 'live' }, + ) => void /** * Callback when a custom event is received from a server-side tool. diff --git a/packages/ai-client/tests/chat-client-interrupts.test.ts b/packages/ai-client/tests/chat-client-interrupts.test.ts index 89f27e086f..50add0a5b4 100644 --- a/packages/ai-client/tests/chat-client-interrupts.test.ts +++ b/packages/ai-client/tests/chat-client-interrupts.test.ts @@ -1099,11 +1099,46 @@ describe('ChatClient native interrupts', () => { expect(onInterruptStateChange).toHaveBeenLastCalledWith( client.getInterruptState(), + { source: 'hydrate' }, ) const state = onInterruptStateChange.mock.lastCall?.[0] expect(state?.interrupts).toBe(state?.pendingInterrupts) }) + it('reports source hydrate when a client-tool interrupt is restored', () => { + const onInterruptStateChange = vi.fn() + const outputSchemaHash = hashSchemaInput(lookupDefinition.outputSchema) + const client = new ChatClient({ + connection: { async *connect() {} }, + tools, + onInterruptStateChange, + initialResumeSnapshot: { + resumeState: { threadId: 'thread-1', runId: 'run-1' }, + pendingInterrupts: [ + descriptor({ + v: INTERRUPT_BINDING_VERSION, + kind: 'client-tool-execution', + interruptId: 'client-1', + interruptedRunId: 'run-1', + generation: 1, + toolName: 'lookup', + toolCallId: 'call-2', + outputSchemaHash, + responseSchemaHash: outputSchemaHash, + }), + ], + }, + }) + + // Restored client-tool items stay internal, but their hydration still + // publishes the state-change callback that lets an app recover the batch. + expect(client.getInterrupts()).toEqual([]) + expect(onInterruptStateChange).toHaveBeenLastCalledWith( + client.getInterruptState(), + { source: 'hydrate' }, + ) + }) + it('owns one immutable interrupt state and resumes with a fresh child run', async () => { const contexts: Array = [] const sentMessages: Array | Array> = [] @@ -1152,9 +1187,18 @@ describe('ChatClient native interrupts', () => { } }, } - const client = new ChatClient({ connection, threadId: 'thread-1' }) + const onInterruptStateChange = vi.fn() + const client = new ChatClient({ + connection, + threadId: 'thread-1', + onInterruptStateChange, + }) await client.sendMessage('start') + expect(onInterruptStateChange).toHaveBeenLastCalledWith( + client.getInterruptState(), + { source: 'live' }, + ) const state = client.getInterruptState() expect(Object.isFrozen(state)).toBe(true) expect(state.interrupts).toBe(state.pendingInterrupts) diff --git a/packages/ai-client/tests/interrupts-types.test-d.ts b/packages/ai-client/tests/interrupts-types.test-d.ts index e025102521..37404173ee 100644 --- a/packages/ai-client/tests/interrupts-types.test-d.ts +++ b/packages/ai-client/tests/interrupts-types.test-d.ts @@ -146,8 +146,9 @@ expectTypeOf>().toEqualTypeOf< declare const client: ChatClient client.updateOptions({ - onInterruptStateChange: (state) => { + onInterruptStateChange: (state, context) => { expectTypeOf(state.interrupts).toEqualTypeOf(state.pendingInterrupts) + expectTypeOf(context.source).toEqualTypeOf<'hydrate' | 'live'>() }, }) client.resolveInterrupts((interrupt) => { diff --git a/packages/ai-client/tests/resume-snapshot.test.ts b/packages/ai-client/tests/resume-snapshot.test.ts index 4bc0ecb5ad..2d73c7ac61 100644 --- a/packages/ai-client/tests/resume-snapshot.test.ts +++ b/packages/ai-client/tests/resume-snapshot.test.ts @@ -1,5 +1,11 @@ import { describe, expect, it, vi } from 'vitest' -import { INTERRUPT_BINDING_VERSION } from '@tanstack/ai/client' +import { + EventType, + INTERRUPT_BINDING_VERSION, + hashSchemaInput, + toolDefinition, +} from '@tanstack/ai/client' +import { z } from 'zod' import { ChatPersistor } from '../src/client-persistor' import { normalizeConnectionAdapter } from '../src/connection-adapters' import { ChatClient } from '../src/chat-client' @@ -370,6 +376,96 @@ describe('ChatClient auto-rejoin after reload', () => { expect(joinRun).not.toHaveBeenCalled() }) + it('publishes hydrate once before live client-tool cancellation', async () => { + const outputSchema = z.object({ accountId: z.string() }) + const execute = vi.fn(async () => ({ accountId: 'account-1' })) + const lookup = toolDefinition({ + name: 'lookup', + description: 'Look up an account', + outputSchema, + }).client(execute) + const outputSchemaHash = hashSchemaInput(outputSchema) + const contexts: Array = [] + const connection: ResumableConnectConnectionAdapter = { + async *connect(_messages, _data, _signal, context) { + contexts.push(context) + const runId = context?.runId ?? 'run-cancelled' + const threadId = context?.threadId ?? 't1' + yield { + type: EventType.RUN_STARTED, + runId, + threadId, + timestamp: Date.now(), + } + yield { + type: EventType.RUN_FINISHED, + runId, + threadId, + timestamp: Date.now(), + outcome: { type: 'success' }, + } + }, + joinRun: async function* () {}, + hydrate: () => + Promise.resolve({ + messages: [createUIMessage('u1', 'look up the account', 'user')], + activeRun: null, + interrupts: { + runId: 'run-paused', + pending: [ + { + id: 'client-1', + reason: 'tanstack:client_tool_execution', + toolCallId: 'call-1', + metadata: { + 'tanstack:interruptBinding': { + v: INTERRUPT_BINDING_VERSION, + kind: 'client-tool-execution', + interruptId: 'client-1', + interruptedRunId: 'run-paused', + generation: 1, + toolName: 'lookup', + toolCallId: 'call-1', + outputSchemaHash, + responseSchemaHash: outputSchemaHash, + }, + }, + }, + ], + }, + }), + } + const sources: Array<'hydrate' | 'live'> = [] + let client: ChatClient + client = mountedChatClient({ + threadId: 't1', + connection, + persistence: true, + tools: [lookup], + onInterruptStateChange: (_state, context) => { + sources.push(context.source) + // Cancelling here re-enters the manager; nested publications must be + // live instead of inheriting the outer hydration source. + if (context.source === 'hydrate') client.cancelInterrupts() + }, + }) + + await vi.waitFor(() => expect(contexts).toHaveLength(1)) + await vi.waitFor(() => expect(client.getResumeState()).toBeNull()) + + expect(sources[0]).toBe('hydrate') + expect(sources.filter((source) => source === 'hydrate')).toEqual([ + 'hydrate', + ]) + expect(sources).toContain('live') + expect(contexts[0]?.parentRunId).toBe('run-paused') + expect(contexts[0]?.resume).toEqual([ + { interruptId: 'client-1', status: 'cancelled' }, + ]) + expect(contexts).toHaveLength(1) + expect(execute).not.toHaveBeenCalled() + }) + it('restores a pending interrupt even when hydrate also reports an activeRun', async () => { // A run paused on an interrupt can momentarily still read as `running` on the // server (the status settles just after the interrupt is persisted), so a diff --git a/packages/ai-preact/src/use-chat.ts b/packages/ai-preact/src/use-chat.ts index bd50d1180d..4a34b41486 100644 --- a/packages/ai-preact/src/use-chat.ts +++ b/packages/ai-preact/src/use-chat.ts @@ -93,7 +93,7 @@ export function useChat< messagesRef.current = messages }, [messages]) - const client = useMemo(() => { + const { client, initializationCallbacks } = useMemo(() => { const messagesToUse = options.initialMessages || [] isFirstMountRef.current = false @@ -116,7 +116,19 @@ export function useChat< } return currentInstance } - const pendingInitializationErrors: Array = [] + // ChatClient may publish while its constructor is still running. Preserve + // those exact notifications until this render commits; invoking them here + // would run state setters and user callbacks during render. + const initializationCallbacks: Array<() => void> = [] + const runOrQueueForActiveInstance = (callback: () => void) => { + const currentInstance = instanceHolder.current + if (!currentInstance) { + initializationCallbacks.push(callback) + return + } + if (activeClientRef.current !== currentInstance) return + callback() + } const instance = new ChatClient({ devtoolsBridgeFactory: createChatDevtoolsBridge, ...transport, @@ -159,13 +171,9 @@ export function useChat< optionsRef.current.onFinish?.(message) }, onError: (err) => { - const currentInstance = instanceHolder.current - if (!currentInstance) { - pendingInitializationErrors.push(err) - return - } - if (activeClientRef.current !== currentInstance) return - optionsRef.current.onError?.(err) + runOrQueueForActiveInstance(() => { + optionsRef.current.onError?.(err) + }) }, onCustomEvent: (eventType, data, context) => { if (!getActiveInstance()) return @@ -226,19 +234,19 @@ export function useChat< pendingInterrupts: nextPendingInterrupts, })) }, - onInterruptStateChange: (nextInterruptState) => { - if (!getActiveInstance()) return - setInterruptState(nextInterruptState) - optionsRef.current.onInterruptStateChange?.(nextInterruptState) + onInterruptStateChange: (nextInterruptState, context) => { + runOrQueueForActiveInstance(() => { + setInterruptState(nextInterruptState) + optionsRef.current.onInterruptStateChange?.( + nextInterruptState, + context, + ) + }) }, }) instanceHolder.current = instance activeClientRef.current = instance - for (const initializationError of pendingInitializationErrors) { - if (activeClientRef.current !== instance) break - optionsRef.current.onError?.(initializationError) - } - return instance + return { client: instance, initializationCallbacks } }, [clientId, syncResumeState]) useEffect(() => { @@ -308,6 +316,10 @@ export function useChat< cleanupInvalidationRef.current = null } activeClientRef.current = client + for (const callback of initializationCallbacks.splice(0)) { + if (activeClientRef.current !== client) break + callback() + } client.mountDevtools() // Delivery-durability resume is transparent: the resumable SSE connection // adapter reattaches via the browser's native Last-Event-ID on reconnect. @@ -341,7 +353,7 @@ export function useChat< } cleanupDisposalRef.current = disposal } - }, [client, syncResumeState]) + }, [client, initializationCallbacks, syncResumeState]) // All callback options are read through optionsRef at call time, so fresh // closures from each render are picked up without recreating the client. diff --git a/packages/ai-preact/tests/use-chat.test.ts b/packages/ai-preact/tests/use-chat.test.ts index 149a592241..0fad1a234c 100644 --- a/packages/ai-preact/tests/use-chat.test.ts +++ b/packages/ai-preact/tests/use-chat.test.ts @@ -38,6 +38,11 @@ describe('useChat', () => { onInterruptStateChange, }) + expect(onInterruptStateChange).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ interrupts: result.current.interrupts }), + { source: 'hydrate' }, + ) expect(Object.isFrozen(result.current.interrupts)).toBe(true) expect(result.current.pendingInterrupts).toBe(result.current.interrupts) expect(result.current.interrupts[0]).toMatchObject({ @@ -69,6 +74,7 @@ describe('useChat', () => { interrupts: result.current.interrupts, interruptErrors: result.current.interruptErrors, }), + { source: 'live' }, ) }) @@ -180,6 +186,32 @@ describe('useChat', () => { expect(persistence.getItem).toHaveBeenCalledWith('persisted-chat') }) + it('should forward synchronous persisted interrupt hydration', () => { + const onInterruptStateChange = vi.fn() + const persistence = { + getItem: vi.fn(() => ({ + messages: [], + resume: createInterruptResumeSnapshot(), + })), + setItem: vi.fn(), + removeItem: vi.fn(), + } + + const { result } = renderUseChat({ + connection: createMockConnectionAdapter(), + threadId: 'persisted-interrupt-chat', + persistence, + onInterruptStateChange, + }) + + expect(result.current.interrupts).toHaveLength(2) + expect(onInterruptStateChange).toHaveBeenCalledOnce() + expect(onInterruptStateChange).toHaveBeenCalledWith( + expect.objectContaining({ interrupts: result.current.interrupts }), + { source: 'hydrate' }, + ) + }) + it('should preserve persisted empty messages over provided initial messages', async () => { const adapter = createMockConnectionAdapter() const initialMessages: Array = [ diff --git a/packages/ai-react/src/use-chat.ts b/packages/ai-react/src/use-chat.ts index 96a75eb9ca..5bcc2d81cd 100644 --- a/packages/ai-react/src/use-chat.ts +++ b/packages/ai-react/src/use-chat.ts @@ -99,7 +99,7 @@ export function useChat< }, []) // Create ChatClient instance with callbacks to sync state - const client = useMemo(() => { + const { client, initializationCallbacks } = useMemo(() => { const messagesToUse = options.initialMessages || [] isFirstMountRef.current = false @@ -122,7 +122,19 @@ export function useChat< } return currentInstance } - const pendingInitializationErrors: Array = [] + // ChatClient may publish while its constructor is still running. Preserve + // those exact notifications until this render commits; invoking them here + // would run state setters and user callbacks during render. + const initializationCallbacks: Array<() => void> = [] + const runOrQueueForActiveInstance = (callback: () => void) => { + const currentInstance = instanceHolder.current + if (!currentInstance) { + initializationCallbacks.push(callback) + return + } + if (activeClientRef.current !== currentInstance) return + callback() + } const instance = new ChatClient({ devtoolsBridgeFactory: createChatDevtoolsBridge, ...transport, @@ -162,13 +174,9 @@ export function useChat< optionsRef.current.onFinish?.(message) }, onError: (error: Error) => { - const currentInstance = instanceHolder.current - if (!currentInstance) { - pendingInitializationErrors.push(error) - return - } - if (activeClientRef.current !== currentInstance) return - optionsRef.current.onError?.(error) + runOrQueueForActiveInstance(() => { + optionsRef.current.onError?.(error) + }) }, ...(initialOptions.tools !== undefined && { tools: initialOptions.tools, @@ -229,19 +237,19 @@ export function useChat< pendingInterrupts: nextPendingInterrupts, })) }, - onInterruptStateChange: (nextInterruptState) => { - if (!getActiveInstance()) return - setInterruptState(nextInterruptState) - optionsRef.current.onInterruptStateChange?.(nextInterruptState) + onInterruptStateChange: (nextInterruptState, context) => { + runOrQueueForActiveInstance(() => { + setInterruptState(nextInterruptState) + optionsRef.current.onInterruptStateChange?.( + nextInterruptState, + context, + ) + }) }, }) instanceHolder.current = instance activeClientRef.current = instance - for (const error of pendingInitializationErrors) { - if (activeClientRef.current !== instance) break - optionsRef.current.onError?.(error) - } - return instance + return { client: instance, initializationCallbacks } }, [clientId, syncResumeState]) useEffect(() => { @@ -328,6 +336,10 @@ export function useChat< cleanupInvalidationRef.current = null } activeClientRef.current = client + for (const callback of initializationCallbacks.splice(0)) { + if (activeClientRef.current !== client) break + callback() + } client.mountDevtools() // Delivery-durability resume is transparent: the resumable SSE connection // adapter re-attaches via the browser's native Last-Event-ID on reconnect. @@ -365,7 +377,7 @@ export function useChat< } cleanupDisposalRef.current = disposal } - }, [client, syncResumeState]) + }, [client, initializationCallbacks, syncResumeState]) const sendMessage = useCallback( async ( diff --git a/packages/ai-react/tests/use-chat.test.ts b/packages/ai-react/tests/use-chat.test.ts index d172ab858a..c8953439a6 100644 --- a/packages/ai-react/tests/use-chat.test.ts +++ b/packages/ai-react/tests/use-chat.test.ts @@ -36,12 +36,21 @@ describe('useChat', () => { describe('interrupt state', () => { it('projects one immutable snapshot with the deprecated pending alias', async () => { const onInterruptStateChange = vi.fn() - const { result } = renderUseChat({ - connection: createMockConnectionAdapter(), - initialResumeSnapshot: createInterruptResumeSnapshot(), - onInterruptStateChange, - }) + const { result } = renderHook( + () => + useChat({ + connection: createMockConnectionAdapter(), + initialResumeSnapshot: createInterruptResumeSnapshot(), + onInterruptStateChange, + }), + { wrapper: StrictMode }, + ) + expect(onInterruptStateChange).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ interrupts: result.current.interrupts }), + { source: 'hydrate' }, + ) expect(Object.isFrozen(result.current.interrupts)).toBe(true) expect(result.current.pendingInterrupts).toBe(result.current.interrupts) expect(result.current.interrupts[0]).toMatchObject({ @@ -73,6 +82,7 @@ describe('useChat', () => { interrupts: result.current.interrupts, interruptErrors: result.current.interruptErrors, }), + { source: 'live' }, ) }) @@ -184,6 +194,36 @@ describe('useChat', () => { expect(persistence.getItem).toHaveBeenCalledWith('persisted-chat') }) + it('should forward synchronous persisted interrupt hydration', () => { + const onInterruptStateChange = vi.fn() + const persistence = { + getItem: vi.fn(() => ({ + messages: [], + resume: createInterruptResumeSnapshot(), + })), + setItem: vi.fn(), + removeItem: vi.fn(), + } + + const { result } = renderHook( + () => + useChat({ + connection: createMockConnectionAdapter(), + threadId: 'persisted-interrupt-chat', + persistence, + onInterruptStateChange, + }), + { wrapper: StrictMode }, + ) + + expect(result.current.interrupts).toHaveLength(2) + expect(onInterruptStateChange).toHaveBeenCalledOnce() + expect(onInterruptStateChange).toHaveBeenCalledWith( + expect.objectContaining({ interrupts: result.current.interrupts }), + { source: 'hydrate' }, + ) + }) + it('should preserve persisted empty messages over provided initial messages', async () => { const adapter = createMockConnectionAdapter() const initialMessages: Array = [ diff --git a/packages/ai-solid/src/use-chat.ts b/packages/ai-solid/src/use-chat.ts index 6ac73e332a..cf90b1b3c8 100644 --- a/packages/ai-solid/src/use-chat.ts +++ b/packages/ai-solid/src/use-chat.ts @@ -180,9 +180,9 @@ export function useChat< pendingInterrupts: nextPendingInterrupts, })) }, - onInterruptStateChange: (nextInterruptState) => { + onInterruptStateChange: (nextInterruptState, context) => { setInterruptState(nextInterruptState) - options.onInterruptStateChange?.(nextInterruptState) + options.onInterruptStateChange?.(nextInterruptState, context) }, }) // Only recreate when clientId changes diff --git a/packages/ai-solid/tests/use-chat.test.ts b/packages/ai-solid/tests/use-chat.test.ts index fe0fcd48b1..492ae85e6e 100644 --- a/packages/ai-solid/tests/use-chat.test.ts +++ b/packages/ai-solid/tests/use-chat.test.ts @@ -56,6 +56,7 @@ describe('useChat', () => { interrupts: chat.interrupts(), interruptErrors: chat.interruptErrors(), }), + { source: 'live' }, ) }) diff --git a/packages/ai-svelte/src/create-chat.svelte.ts b/packages/ai-svelte/src/create-chat.svelte.ts index 873010503f..90ba4d404d 100644 --- a/packages/ai-svelte/src/create-chat.svelte.ts +++ b/packages/ai-svelte/src/create-chat.svelte.ts @@ -179,9 +179,9 @@ export function createChat< onRunIdChange: (nextRunId) => { runId = nextRunId }, - onInterruptStateChange: (nextInterruptState) => { + onInterruptStateChange: (nextInterruptState, context) => { interruptState = nextInterruptState - options.onInterruptStateChange?.(nextInterruptState) + options.onInterruptStateChange?.(nextInterruptState, context) }, }) diff --git a/packages/ai-svelte/tests/use-chat.test.ts b/packages/ai-svelte/tests/use-chat.test.ts index e9ee4b732a..297826ef36 100644 --- a/packages/ai-svelte/tests/use-chat.test.ts +++ b/packages/ai-svelte/tests/use-chat.test.ts @@ -54,6 +54,7 @@ describe('createChat', () => { interrupts: chat.interrupts, interruptErrors: chat.interruptErrors, }), + { source: 'live' }, ) }) diff --git a/packages/ai-vue/src/use-chat.ts b/packages/ai-vue/src/use-chat.ts index fe5c9e4514..d9b8a91e03 100644 --- a/packages/ai-vue/src/use-chat.ts +++ b/packages/ai-vue/src/use-chat.ts @@ -164,9 +164,9 @@ export function useChat< onRunIdChange: (nextRunId) => { runId.value = nextRunId }, - onInterruptStateChange: (nextInterruptState) => { + onInterruptStateChange: (nextInterruptState, context) => { interruptState.value = nextInterruptState - options.onInterruptStateChange?.(nextInterruptState) + options.onInterruptStateChange?.(nextInterruptState, context) }, }) diff --git a/packages/ai-vue/tests/use-chat.test.ts b/packages/ai-vue/tests/use-chat.test.ts index 6e92f4b259..65a019ee79 100644 --- a/packages/ai-vue/tests/use-chat.test.ts +++ b/packages/ai-vue/tests/use-chat.test.ts @@ -55,6 +55,7 @@ describe('useChat', () => { interrupts: result.current.interrupts, interruptErrors: result.current.interruptErrors, }), + { source: 'live' }, ) }) diff --git a/testing/e2e/src/routes/persistence-durability.tsx b/testing/e2e/src/routes/persistence-durability.tsx index 8f7aae0a76..6394f1c6e4 100644 --- a/testing/e2e/src/routes/persistence-durability.tsx +++ b/testing/e2e/src/routes/persistence-durability.tsx @@ -53,6 +53,9 @@ export const Route = createFileRoute('/persistence-durability')({ function PersistenceDurabilityPage() { const { scenario } = Route.useSearch() + const [interruptSource, setInterruptSource] = useState< + 'hydrate' | 'live' | null + >(null) const isInterrupt = scenario === 'interrupt' const isServerInterrupt = scenario === 'server-interrupt' const chatId = isServerInterrupt @@ -72,6 +75,9 @@ function PersistenceDurabilityPage() { ? interruptConnection : textConnection, persistence: isServerInterrupt ? true : store, + onInterruptStateChange: (_state, context) => { + setInterruptSource(context.source) + }, }) const [input, setInput] = useState('') @@ -95,6 +101,9 @@ function PersistenceDurabilityPage() { data-count={String(messages.length)} hidden /> +
{messages.map((message) => ( diff --git a/testing/e2e/tests/persistence-durability.spec.ts b/testing/e2e/tests/persistence-durability.spec.ts index 550debd0e3..2619603285 100644 --- a/testing/e2e/tests/persistence-durability.spec.ts +++ b/testing/e2e/tests/persistence-durability.spec.ts @@ -85,6 +85,7 @@ test.describe('persistence durability (browser refresh)', () => { .poll(() => interruptCount(page), { timeout: 15_000 }) .toBeGreaterThanOrEqual(1) await expect(page.getByTestId('interrupt-confirm-shipment')).toBeVisible() + await expect(page.getByTestId('interrupt-source')).toHaveText('live') // The combined record carries the resume half while the interrupt is pending. const stored = await page.evaluate(() => @@ -105,6 +106,7 @@ test.describe('persistence durability (browser refresh)', () => { .toBeGreaterThanOrEqual(1) await expect(page.getByTestId('interrupt-confirm-shipment')).toBeVisible() await expect(page.getByTestId('interrupt-kind')).toHaveText('generic') + await expect(page.getByTestId('interrupt-source')).toHaveText('hydrate') }) test('restores a pending interrupt from the SERVER on a fresh load (persistence: true)', async ({ @@ -126,6 +128,7 @@ test.describe('persistence durability (browser refresh)', () => { .toBeGreaterThanOrEqual(1) await expect(page.getByTestId('interrupt-confirm-shipment')).toBeVisible() await expect(page.getByTestId('interrupt-kind')).toHaveText('generic') + await expect(page.getByTestId('interrupt-source')).toHaveText('hydrate') // Restored bound and resolvable — the reload can approve/reject, not just // view a dead paused tool call. await expect(page.getByTestId('interrupt-can-resolve')).toHaveText('true') From 3f9ae0f69b2e1328b3693e016b33adf189112be4 Mon Sep 17 00:00:00 2001 From: kolaworld Date: Sun, 16 Aug 2026 13:00:04 -0400 Subject: [PATCH 2/6] docs(ai): document restored interrupt policies --- docs/api/ai-angular.md | 1 + docs/api/ai-client.md | 1 + docs/api/ai-preact.md | 1 + docs/api/ai-react.md | 1 + docs/api/ai-solid.md | 1 + docs/api/ai-svelte.md | 1 + docs/api/ai-vue.md | 1 + docs/config.json | 25 ++++++++------- docs/interrupts/multiple.md | 7 +++- docs/interrupts/overview.md | 5 +++ docs/persistence/client-persistence.md | 44 ++++++++++++++++++++++++++ docs/tools/client-tools.md | 6 ++-- 12 files changed, 80 insertions(+), 14 deletions(-) diff --git a/docs/api/ai-angular.md b/docs/api/ai-angular.md index 2ae10c6589..fd1391a3d0 100644 --- a/docs/api/ai-angular.md +++ b/docs/api/ai-angular.md @@ -65,6 +65,7 @@ Extends `ChatClientOptions` from `@tanstack/ai-client` (minus internal state cal - `onChunk?` - Callback when stream chunk is received - `onFinish?` - Callback when response finishes - `onError?` - Callback when error occurs +- `onInterruptStateChange?` - Callback when interrupt state changes; context source is `hydrate` for restored state and `live` for streamed or client-initiated updates - `onCustomEvent?` - Callback for custom stream events - `streamProcessor?` - Stream processing configuration diff --git a/docs/api/ai-client.md b/docs/api/ai-client.md index fda57f7e7f..b6e3410e1b 100644 --- a/docs/api/ai-client.md +++ b/docs/api/ai-client.md @@ -105,6 +105,7 @@ goes away. Users of the framework hooks need no change. - `onChunk?` - Callback when stream chunk is received - `onFinish?` - Callback when response finishes - `onError?` - Callback when error occurs +- `onInterruptStateChange?` - Callback when interrupt state changes; context source is `hydrate` for restored state and `live` for streamed or client-initiated updates - `onMessagesChange?` - Callback when messages change - `onLoadingChange?` - Callback when loading state changes - `onErrorChange?` - Callback when error state changes diff --git a/docs/api/ai-preact.md b/docs/api/ai-preact.md index 8876a6e628..e63dfe3480 100644 --- a/docs/api/ai-preact.md +++ b/docs/api/ai-preact.md @@ -82,6 +82,7 @@ Extends `ChatClientOptions` from `@tanstack/ai-client`: - `onChunk?` - Callback when stream chunk is received - `onFinish?` - Callback when response finishes - `onError?` - Callback when error occurs +- `onInterruptStateChange?` - Callback when interrupt state changes; context source is `hydrate` for restored state and `live` for streamed or client-initiated updates - `streamProcessor?` - Stream processing configuration **Note:** Client tools are now automatically executed - no `onToolCall` callback needed! diff --git a/docs/api/ai-react.md b/docs/api/ai-react.md index 01eb1587b1..3e10cebe23 100644 --- a/docs/api/ai-react.md +++ b/docs/api/ai-react.md @@ -92,6 +92,7 @@ Extends `ChatClientOptions` from `@tanstack/ai-client`: - `onChunk?` - Callback when stream chunk is received - `onFinish?` - Callback when response finishes - `onError?` - Callback when error occurs +- `onInterruptStateChange?` - Callback when interrupt state changes; context source is `hydrate` for restored state and `live` for streamed or client-initiated updates - `streamProcessor?` - Stream processing configuration **Note:** Client tools are now automatically executed - no `onToolCall` callback needed! diff --git a/docs/api/ai-solid.md b/docs/api/ai-solid.md index 46de176a99..8b2b7352f4 100644 --- a/docs/api/ai-solid.md +++ b/docs/api/ai-solid.md @@ -83,6 +83,7 @@ Extends `ChatClientOptions` from `@tanstack/ai-client`: - `onChunk?` - Callback when stream chunk is received - `onFinish?` - Callback when response finishes - `onError?` - Callback when error occurs +- `onInterruptStateChange?` - Callback when interrupt state changes; context source is `hydrate` for restored state and `live` for streamed or client-initiated updates - `streamProcessor?` - Stream processing configuration **Note:** Client tools are now automatically executed - no `onToolCall` callback needed! diff --git a/docs/api/ai-svelte.md b/docs/api/ai-svelte.md index 1792215949..9cea04ed62 100644 --- a/docs/api/ai-svelte.md +++ b/docs/api/ai-svelte.md @@ -82,6 +82,7 @@ Extends `ChatClientOptions` from `@tanstack/ai-client` (minus internal state cal - `onChunk?` - Callback when stream chunk is received - `onFinish?` - Callback when response finishes - `onError?` - Callback when error occurs +- `onInterruptStateChange?` - Callback when interrupt state changes; context source is `hydrate` for restored state and `live` for streamed or client-initiated updates - `onCustomEvent?` - Callback for custom stream events - `streamProcessor?` - Stream processing configuration diff --git a/docs/api/ai-vue.md b/docs/api/ai-vue.md index 7358000cb9..499d812ce6 100644 --- a/docs/api/ai-vue.md +++ b/docs/api/ai-vue.md @@ -80,6 +80,7 @@ Extends `ChatClientOptions` from `@tanstack/ai-client` (minus internal state cal - `onChunk?` - Callback when stream chunk is received - `onFinish?` - Callback when response finishes - `onError?` - Callback when error occurs +- `onInterruptStateChange?` - Callback when interrupt state changes; context source is `hydrate` for restored state and `live` for streamed or client-initiated updates - `onCustomEvent?` - Callback for custom stream events - `streamProcessor?` - Stream processing configuration diff --git a/docs/config.json b/docs/config.json index e5c98bc877..7c5b0408da 100644 --- a/docs/config.json +++ b/docs/config.json @@ -105,7 +105,7 @@ "label": "Client Tools", "to": "tools/client-tools", "addedAt": "2026-04-15", - "updatedAt": "2026-07-21" + "updatedAt": "2026-08-16" }, { "label": "Tool Approval Flow", @@ -188,7 +188,8 @@ { "label": "Overview", "to": "interrupts/overview", - "addedAt": "2026-08-04" + "addedAt": "2026-08-04", + "updatedAt": "2026-08-16" }, { "label": "Tool Approval", @@ -198,7 +199,8 @@ { "label": "Multiple Interrupts", "to": "interrupts/multiple", - "addedAt": "2026-08-04" + "addedAt": "2026-08-04", + "updatedAt": "2026-08-16" }, { "label": "Generic Interrupts", @@ -250,7 +252,8 @@ { "label": "Client Persistence", "to": "persistence/client-persistence", - "addedAt": "2026-08-04" + "addedAt": "2026-08-04", + "updatedAt": "2026-08-16" }, { "label": "Generation Persistence", @@ -737,43 +740,43 @@ "label": "@tanstack/ai-client", "to": "api/ai-client", "addedAt": "2026-04-15", - "updatedAt": "2026-08-04" + "updatedAt": "2026-08-16" }, { "label": "@tanstack/ai-react", "to": "api/ai-react", "addedAt": "2026-04-15", - "updatedAt": "2026-07-08" + "updatedAt": "2026-08-16" }, { "label": "@tanstack/ai-solid", "to": "api/ai-solid", "addedAt": "2026-04-15", - "updatedAt": "2026-07-08" + "updatedAt": "2026-08-16" }, { "label": "@tanstack/ai-preact", "to": "api/ai-preact", "addedAt": "2026-04-15", - "updatedAt": "2026-07-08" + "updatedAt": "2026-08-16" }, { "label": "@tanstack/ai-vue", "to": "api/ai-vue", "addedAt": "2026-04-15", - "updatedAt": "2026-07-30" + "updatedAt": "2026-08-16" }, { "label": "@tanstack/ai-svelte", "to": "api/ai-svelte", "addedAt": "2026-04-15", - "updatedAt": "2026-07-30" + "updatedAt": "2026-08-16" }, { "label": "@tanstack/ai-angular", "to": "api/ai-angular", "addedAt": "2026-06-15", - "updatedAt": "2026-07-30" + "updatedAt": "2026-08-16" } ] }, diff --git a/docs/interrupts/multiple.md b/docs/interrupts/multiple.md index 0b0ebd78b4..cc9286b4d5 100644 --- a/docs/interrupts/multiple.md +++ b/docs/interrupts/multiple.md @@ -130,7 +130,12 @@ Two shortcuts cover the common cases: whole queue. It works only when every item is a tool approval that needs no payload or edits. Generic items, mixed queues, or required payloads are rejected. -- `cancelInterrupts()` cancels every item with no payload. +- `cancelInterrupts()` cancels the complete internal batch with no payload, + including client-tool execution steps that are hidden from `interrupts`. + +Use the `onInterruptStateChange` source to apply different policies to restored +(`hydrate`) and current-session (`live`) batches. Cancellation still applies to +the complete batch. ## When an answer is wrong diff --git a/docs/interrupts/overview.md b/docs/interrupts/overview.md index 9a5233b106..88a1f23017 100644 --- a/docs/interrupts/overview.md +++ b/docs/interrupts/overview.md @@ -183,6 +183,11 @@ A tool with a `.client()` implementation runs in the browser on its own and reports its own result. That is not a decision you make, so it never appears in `interrupts`. See [Client Tools](../tools/client-tools). +If persistence restores a pending client-tool execution, the client leaves it +pending rather than running the browser code again. See +[Client persistence](../persistence/client-persistence#handle-restored-client-tools) +for recovery policies. + The one time a tool pauses is when you mark it `needsApproval: true`. Then it stops for a yes or no first, whether it runs on the server or in the browser: diff --git a/docs/persistence/client-persistence.md b/docs/persistence/client-persistence.md index f2d8199cb3..b38673ea1c 100644 --- a/docs/persistence/client-persistence.md +++ b/docs/persistence/client-persistence.md @@ -67,6 +67,50 @@ pointer. On the next load `useChat` reads it and: a durability-backed connection (a route that records the stream and exposes a replay handler); see [Resumable streams](../resumable-streams/overview). +### Handle restored client tools + +A live client tool runs automatically when its call arrives from the stream. +Hydration restores its pending execution but does not run the browser code +again, because that work may not be safe to repeat. The execution remains an +internal interrupt and does not appear in `interrupts`. + +Use `onInterruptStateChange` to distinguish restored interrupt state from live +updates. Leaving a restored batch pending is the default. This example cancels +every restored batch instead: + +```tsx +import { useEffect, useState } from 'react' +import { + fetchServerSentEvents, + localStoragePersistence, + useChat, +} from '@tanstack/ai-react' + +function Chat() { + const [restoredBatch, setRestoredBatch] = useState(false) + const { cancelInterrupts } = useChat({ + threadId: 'support-chat', + connection: fetchServerSentEvents('/api/chat'), + persistence: localStoragePersistence(), + onInterruptStateChange(_state, { source }) { + setRestoredBatch(source === 'hydrate') + }, + }) + + useEffect(() => { + if (restoredBatch) cancelInterrupts() + }, [cancelInterrupts, restoredBatch]) + + return null +} +``` + +`source` is `hydrate` for state restored from an initial resume snapshot, a +client storage adapter, or server hydration. It is `live` for streamed and +client-initiated changes. `cancelInterrupts()` cancels the complete internal +batch, including visible approvals and hidden client-tool executions; it does +not selectively cancel one kind of interrupt. + ## Choose a mode `persistence` takes a storage adapter or a boolean: diff --git a/docs/tools/client-tools.md b/docs/tools/client-tools.md index 8f21dc04a6..ffcee5c2ee 100644 --- a/docs/tools/client-tools.md +++ b/docs/tools/client-tools.md @@ -65,8 +65,10 @@ Native client-tool execution shares the atomic interrupt **batch** lifecycle (it can gate multi-item submits) but is **auto-resolved** — you do not call `resolveInterrupt` for it. See [Interrupts](../interrupts/overview) for the ephemeral lifecycle, batches, and migration from the historical -`tool-input-available` custom event. Durable recovery is optional and not part -of the default client-tool path. +`tool-input-available` custom event. After hydration, a pending client-tool +execution is restored but not run again. See +[Client persistence](../persistence/client-persistence#handle-restored-client-tools) +for recovery policies. ## Approval is a separate axis From 2514d9643aa399a8cf360bae024759a6f55b275a Mon Sep 17 00:00:00 2001 From: kolaworld Date: Sun, 16 Aug 2026 17:47:21 -0400 Subject: [PATCH 3/6] fix(frameworks): preserve initialization callbacks --- packages/ai-preact/src/use-chat.ts | 125 +++++++++++++-------- packages/ai-preact/tests/use-chat.test.ts | 59 ++++++++-- packages/ai-react/src/use-chat.ts | 130 +++++++++++++--------- packages/ai-react/tests/use-chat.test.ts | 47 +++++++- 4 files changed, 251 insertions(+), 110 deletions(-) diff --git a/packages/ai-preact/src/use-chat.ts b/packages/ai-preact/src/use-chat.ts index 4a34b41486..1bdaf09c03 100644 --- a/packages/ai-preact/src/use-chat.ts +++ b/packages/ai-preact/src/use-chat.ts @@ -93,7 +93,7 @@ export function useChat< messagesRef.current = messages }, [messages]) - const { client, initializationCallbacks } = useMemo(() => { + const { client, initialization } = useMemo(() => { const messagesToUse = options.initialMessages || [] isFirstMountRef.current = false @@ -116,17 +116,22 @@ export function useChat< } return currentInstance } - // ChatClient may publish while its constructor is still running. Preserve - // those exact notifications until this render commits; invoking them here - // would run state setters and user callbacks during render. - const initializationCallbacks: Array<() => void> = [] + // ChatClient may publish while its constructor is running or while async + // persistence resolves before commit. Preserve those exact notifications + // until this render commits; invoking them here would run state setters and + // user callbacks for a render that may never mount. + const initializationState = { + ready: false, + callbacks: [] as Array<() => void>, + } const runOrQueueForActiveInstance = (callback: () => void) => { - const currentInstance = instanceHolder.current - if (!currentInstance) { - initializationCallbacks.push(callback) + if (!initializationState.ready) { + initializationState.callbacks.push(callback) return } - if (activeClientRef.current !== currentInstance) return + const currentInstance = instanceHolder.current + if (!currentInstance || activeClientRef.current !== currentInstance) + return callback() } const instance = new ChatClient({ @@ -163,12 +168,14 @@ export function useChat< return optionsRef.current.onResponse?.(response) }, onChunk: (chunk) => { - if (!getActiveInstance()) return - optionsRef.current.onChunk?.(chunk) + runOrQueueForActiveInstance(() => { + optionsRef.current.onChunk?.(chunk) + }) }, onFinish: (message) => { - if (!getActiveInstance()) return - optionsRef.current.onFinish?.(message) + runOrQueueForActiveInstance(() => { + optionsRef.current.onFinish?.(message) + }) }, onError: (err) => { runOrQueueForActiveInstance(() => { @@ -176,8 +183,9 @@ export function useChat< }) }, onCustomEvent: (eventType, data, context) => { - if (!getActiveInstance()) return - optionsRef.current.onCustomEvent?.(eventType, data, context) + runOrQueueForActiveInstance(() => { + optionsRef.current.onCustomEvent?.(eventType, data, context) + }) }, ...(initialOptions.tools !== undefined && { tools: initialOptions.tools, @@ -186,53 +194,64 @@ export function useChat< streamProcessor: options.streamProcessor, }), onMessagesChange: (newMessages: Array>) => { - if (!getActiveInstance()) return - setMessages(newMessages) + runOrQueueForActiveInstance(() => { + setMessages(newMessages) + }) }, onLoadingChange: (newIsLoading: boolean) => { - const currentInstance = getActiveInstance() - if (!currentInstance) return - setIsLoading(newIsLoading) - syncResumeState(currentInstance) + runOrQueueForActiveInstance(() => { + const currentInstance = getActiveInstance() + if (!currentInstance) return + setIsLoading(newIsLoading) + syncResumeState(currentInstance) + }) }, onStatusChange: (newStatus: ChatClientState) => { - if (!getActiveInstance()) return - setStatus(newStatus) + runOrQueueForActiveInstance(() => { + setStatus(newStatus) + }) }, onErrorChange: (newError: Error | undefined) => { - if (!getActiveInstance()) return - setError(newError) + runOrQueueForActiveInstance(() => { + setError(newError) + }) }, onSubscriptionChange: (nextIsSubscribed: boolean) => { - if (!getActiveInstance()) return - setIsSubscribed(nextIsSubscribed) + runOrQueueForActiveInstance(() => { + setIsSubscribed(nextIsSubscribed) + }) }, onConnectionStatusChange: (nextStatus: ConnectionStatus) => { - if (!getActiveInstance()) return - setConnectionStatus(nextStatus) + runOrQueueForActiveInstance(() => { + setConnectionStatus(nextStatus) + }) }, onSessionGeneratingChange: (isGenerating: boolean) => { - if (!getActiveInstance()) return - setSessionGenerating(isGenerating) + runOrQueueForActiveInstance(() => { + setSessionGenerating(isGenerating) + }) }, ...(optionsRef.current.queue !== undefined && { queue: optionsRef.current.queue, }), onQueueChange: (nextQueue: Array) => { - if (!getActiveInstance()) return - setQueue(nextQueue) + runOrQueueForActiveInstance(() => { + setQueue(nextQueue) + }) }, onRunIdChange: (nextRunId) => { - if (!getActiveInstance()) return - setRunId(nextRunId) + runOrQueueForActiveInstance(() => { + setRunId(nextRunId) + }) }, onResumeStateChange: (_nextResumeState, nextPendingInterrupts) => { - if (!getActiveInstance()) return - setInterruptState((current) => ({ - ...current, - interrupts: nextPendingInterrupts, - pendingInterrupts: nextPendingInterrupts, - })) + runOrQueueForActiveInstance(() => { + setInterruptState((current) => ({ + ...current, + interrupts: nextPendingInterrupts, + pendingInterrupts: nextPendingInterrupts, + })) + }) }, onInterruptStateChange: (nextInterruptState, context) => { runOrQueueForActiveInstance(() => { @@ -245,10 +264,23 @@ export function useChat< }, }) instanceHolder.current = instance - activeClientRef.current = instance - return { client: instance, initializationCallbacks } + return { client: instance, initialization: initializationState } }, [clientId, syncResumeState]) + useEffect(() => { + activeClientRef.current = client + // Keep initialization closed while draining so callbacks published by a + // queued callback are appended and delivered in the same commit. + while (initialization.callbacks.length > 0) { + if (activeClientRef.current !== client) { + initialization.callbacks.length = 0 + break + } + initialization.callbacks.shift()?.() + } + initialization.ready = true + }, [client, initialization]) + useEffect(() => { const clientMessages = client.getMessages() if (clientMessages !== messagesRef.current) { @@ -315,11 +347,6 @@ export function useChat< clearTimeout(cleanupInvalidationRef.current) cleanupInvalidationRef.current = null } - activeClientRef.current = client - for (const callback of initializationCallbacks.splice(0)) { - if (activeClientRef.current !== client) break - callback() - } client.mountDevtools() // Delivery-durability resume is transparent: the resumable SSE connection // adapter reattaches via the browser's native Last-Event-ID on reconnect. @@ -353,7 +380,7 @@ export function useChat< } cleanupDisposalRef.current = disposal } - }, [client, initializationCallbacks, syncResumeState]) + }, [client, syncResumeState]) // All callback options are read through optionsRef at call time, so fresh // closures from each render are picked up without recreating the client. diff --git a/packages/ai-preact/tests/use-chat.test.ts b/packages/ai-preact/tests/use-chat.test.ts index 0fad1a234c..e134e503e3 100644 --- a/packages/ai-preact/tests/use-chat.test.ts +++ b/packages/ai-preact/tests/use-chat.test.ts @@ -78,6 +78,40 @@ describe('useChat', () => { ) }) + it('awaits onResponse when hydration resumes during activation', async () => { + const response = createDeferred() + const onResponse = vi.fn(() => response.promise) + const onConnect = vi.fn() + const { result } = renderUseChat({ + connection: createMockConnectionAdapter({ + chunks: createTextChunks('resumed'), + onConnect, + }), + initialResumeSnapshot: createInterruptResumeSnapshot(), + live: true, + onResponse, + onInterruptStateChange: (state, context) => { + if (context.source !== 'hydrate') return + for (const interrupt of state.interrupts) interrupt.cancel() + }, + }) + + await waitFor(() => { + expect(onResponse).toHaveBeenCalledOnce() + }) + expect(onConnect).not.toHaveBeenCalled() + + await act(async () => { + response.resolve() + await response.promise + }) + + await waitFor(() => { + expect(onConnect).toHaveBeenCalledOnce() + expect(result.current.resuming).toBe(false) + }) + }) + it('delegates every root interrupt control to ChatClient', async () => { const resolve = vi .spyOn(ChatClient.prototype, 'resolveInterrupts') @@ -1099,25 +1133,34 @@ describe('useChat', () => { expect(onError.mock.calls[0]?.[0].message).toBe('Test error') }) - it('should call onResponse callback when response is received', async () => { + it('should await onResponse before connecting', async () => { const chunks = createTextChunks('Response') - const adapter = createMockConnectionAdapter({ chunks }) - const onResponse = vi.fn() + const onConnect = vi.fn() + const adapter = createMockConnectionAdapter({ chunks, onConnect }) + const response = createDeferred() + const onResponse = vi.fn(() => response.promise) const { result } = renderUseChat({ connection: adapter, onResponse, }) - await act(async () => { - await result.current.sendMessage('Test') + let sendPromise: Promise + act(() => { + sendPromise = result.current.sendMessage('Test') }) - // onResponse may or may not be called depending on adapter implementation - // This test verifies the callback is passed through await waitFor(() => { - expect(result.current.messages.length).toBeGreaterThan(0) + expect(onResponse).toHaveBeenCalledOnce() }) + expect(onConnect).not.toHaveBeenCalled() + + await act(async () => { + response.resolve() + await sendPromise! + }) + + expect(onConnect).toHaveBeenCalledOnce() }) }) diff --git a/packages/ai-react/src/use-chat.ts b/packages/ai-react/src/use-chat.ts index 5bcc2d81cd..a8ce94812d 100644 --- a/packages/ai-react/src/use-chat.ts +++ b/packages/ai-react/src/use-chat.ts @@ -99,7 +99,7 @@ export function useChat< }, []) // Create ChatClient instance with callbacks to sync state - const { client, initializationCallbacks } = useMemo(() => { + const { client, initialization } = useMemo(() => { const messagesToUse = options.initialMessages || [] isFirstMountRef.current = false @@ -122,17 +122,22 @@ export function useChat< } return currentInstance } - // ChatClient may publish while its constructor is still running. Preserve - // those exact notifications until this render commits; invoking them here - // would run state setters and user callbacks during render. - const initializationCallbacks: Array<() => void> = [] + // ChatClient may publish while its constructor is running or while async + // persistence resolves before commit. Preserve those exact notifications + // until this render commits; invoking them here would run state setters and + // user callbacks for a client React may abandon. + const initializationState = { + ready: false, + callbacks: [] as Array<() => void>, + } const runOrQueueForActiveInstance = (callback: () => void) => { - const currentInstance = instanceHolder.current - if (!currentInstance) { - initializationCallbacks.push(callback) + if (!initializationState.ready) { + initializationState.callbacks.push(callback) return } - if (activeClientRef.current !== currentInstance) return + const currentInstance = instanceHolder.current + if (!currentInstance || activeClientRef.current !== currentInstance) + return callback() } const instance = new ChatClient({ @@ -162,16 +167,19 @@ export function useChat< outputKind: initialOptions.outputSchema ? 'structured' : 'chat', }, onResponse: (response) => { - if (!getActiveInstance()) return - void optionsRef.current.onResponse?.(response) + runOrQueueForActiveInstance(() => { + void optionsRef.current.onResponse?.(response) + }) }, onChunk: (chunk: StreamChunk) => { - if (!getActiveInstance()) return - optionsRef.current.onChunk?.(chunk) + runOrQueueForActiveInstance(() => { + optionsRef.current.onChunk?.(chunk) + }) }, onFinish: (message: UIMessage) => { - if (!getActiveInstance()) return - optionsRef.current.onFinish?.(message) + runOrQueueForActiveInstance(() => { + optionsRef.current.onFinish?.(message) + }) }, onError: (error: Error) => { runOrQueueForActiveInstance(() => { @@ -182,60 +190,72 @@ export function useChat< tools: initialOptions.tools, }), onCustomEvent: (eventType, data, context) => { - if (!getActiveInstance()) return - optionsRef.current.onCustomEvent?.(eventType, data, context) + runOrQueueForActiveInstance(() => { + optionsRef.current.onCustomEvent?.(eventType, data, context) + }) }, ...(options.streamProcessor !== undefined && { streamProcessor: options.streamProcessor, }), onMessagesChange: (newMessages: Array>) => { - if (!getActiveInstance()) return - setMessages(newMessages) + runOrQueueForActiveInstance(() => { + setMessages(newMessages) + }) }, onLoadingChange: (newIsLoading: boolean) => { - const currentInstance = getActiveInstance() - if (!currentInstance) return - setIsLoading(newIsLoading) - syncResumeState(currentInstance) + runOrQueueForActiveInstance(() => { + const currentInstance = getActiveInstance() + if (!currentInstance) return + setIsLoading(newIsLoading) + syncResumeState(currentInstance) + }) }, onErrorChange: (newError: Error | undefined) => { - if (!getActiveInstance()) return - setError(newError) + runOrQueueForActiveInstance(() => { + setError(newError) + }) }, onStatusChange: (status: ChatClientState) => { - if (!getActiveInstance()) return - setStatus(status) + runOrQueueForActiveInstance(() => { + setStatus(status) + }) }, onSubscriptionChange: (nextIsSubscribed: boolean) => { - if (!getActiveInstance()) return - setIsSubscribed(nextIsSubscribed) + runOrQueueForActiveInstance(() => { + setIsSubscribed(nextIsSubscribed) + }) }, onConnectionStatusChange: (nextStatus: ConnectionStatus) => { - if (!getActiveInstance()) return - setConnectionStatus(nextStatus) + runOrQueueForActiveInstance(() => { + setConnectionStatus(nextStatus) + }) }, onSessionGeneratingChange: (isGenerating: boolean) => { - if (!getActiveInstance()) return - setSessionGenerating(isGenerating) + runOrQueueForActiveInstance(() => { + setSessionGenerating(isGenerating) + }) }, ...(optionsRef.current.queue !== undefined && { queue: optionsRef.current.queue, }), onQueueChange: (nextQueue: Array) => { - if (activeClientRef.current !== instance) return - setQueue(nextQueue) + runOrQueueForActiveInstance(() => { + setQueue(nextQueue) + }) }, onRunIdChange: (nextRunId) => { - if (!getActiveInstance()) return - setRunId(nextRunId) + runOrQueueForActiveInstance(() => { + setRunId(nextRunId) + }) }, onResumeStateChange: (_nextResumeState, nextPendingInterrupts) => { - if (!getActiveInstance()) return - setInterruptState((current) => ({ - ...current, - interrupts: nextPendingInterrupts, - pendingInterrupts: nextPendingInterrupts, - })) + runOrQueueForActiveInstance(() => { + setInterruptState((current) => ({ + ...current, + interrupts: nextPendingInterrupts, + pendingInterrupts: nextPendingInterrupts, + })) + }) }, onInterruptStateChange: (nextInterruptState, context) => { runOrQueueForActiveInstance(() => { @@ -248,10 +268,23 @@ export function useChat< }, }) instanceHolder.current = instance - activeClientRef.current = instance - return { client: instance, initializationCallbacks } + return { client: instance, initialization: initializationState } }, [clientId, syncResumeState]) + useEffect(() => { + activeClientRef.current = client + // Keep initialization closed while draining so callbacks published by a + // queued callback are appended and delivered in the same commit. + while (initialization.callbacks.length > 0) { + if (activeClientRef.current !== client) { + initialization.callbacks.length = 0 + break + } + initialization.callbacks.shift()?.() + } + initialization.ready = true + }, [client, initialization]) + useEffect(() => { const clientMessages = client.getMessages() if (clientMessages !== messagesRef.current) { @@ -335,11 +368,6 @@ export function useChat< clearTimeout(cleanupInvalidationRef.current) cleanupInvalidationRef.current = null } - activeClientRef.current = client - for (const callback of initializationCallbacks.splice(0)) { - if (activeClientRef.current !== client) break - callback() - } client.mountDevtools() // Delivery-durability resume is transparent: the resumable SSE connection // adapter re-attaches via the browser's native Last-Event-ID on reconnect. @@ -377,7 +405,7 @@ export function useChat< } cleanupDisposalRef.current = disposal } - }, [client, initializationCallbacks, syncResumeState]) + }, [client, syncResumeState]) const sendMessage = useCallback( async ( diff --git a/packages/ai-react/tests/use-chat.test.ts b/packages/ai-react/tests/use-chat.test.ts index c8953439a6..9921c27baa 100644 --- a/packages/ai-react/tests/use-chat.test.ts +++ b/packages/ai-react/tests/use-chat.test.ts @@ -1,7 +1,7 @@ import { EventType } from '@tanstack/ai' import { ChatClient } from '@tanstack/ai-client' -import { act, renderHook, waitFor } from '@testing-library/react' -import { StrictMode, useState } from 'react' +import { act, render, renderHook, waitFor } from '@testing-library/react' +import { StrictMode, Suspense, createElement, useState } from 'react' import { afterEach, describe, expect, it, vi } from 'vitest' import { useChat } from '../src/use-chat' import { @@ -224,6 +224,49 @@ describe('useChat', () => { ) }) + it('does not publish async hydration from an abandoned render', async () => { + const hydration = createDeferred<{ + messages: Array + resume: ReturnType + }>() + const onInterruptStateChange = vi.fn() + const persistence = { + getItem: vi.fn(() => hydration.promise), + setItem: vi.fn(), + removeItem: vi.fn(), + } + const suspended = new Promise(() => {}) + + function AbandonedChat(): never { + useChat({ + connection: createMockConnectionAdapter(), + threadId: 'abandoned-chat', + persistence, + onInterruptStateChange, + }) + throw suspended + } + + const view = render( + createElement( + Suspense, + { fallback: null }, + createElement(AbandonedChat), + ), + ) + view.unmount() + + await act(async () => { + hydration.resolve({ + messages: [], + resume: createInterruptResumeSnapshot(), + }) + await hydration.promise + }) + + expect(onInterruptStateChange).not.toHaveBeenCalled() + }) + it('should preserve persisted empty messages over provided initial messages', async () => { const adapter = createMockConnectionAdapter() const initialMessages: Array = [ From 0428a96fcc0d7f6b5db34e6443e53f54901f705a Mon Sep 17 00:00:00 2001 From: kolaworld Date: Sun, 16 Aug 2026 22:11:55 -0400 Subject: [PATCH 4/6] docs: remove metadata date bumps from bug fix --- docs/config.json | 25 +++++++++++-------------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/docs/config.json b/docs/config.json index 7c5b0408da..e5c98bc877 100644 --- a/docs/config.json +++ b/docs/config.json @@ -105,7 +105,7 @@ "label": "Client Tools", "to": "tools/client-tools", "addedAt": "2026-04-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-07-21" }, { "label": "Tool Approval Flow", @@ -188,8 +188,7 @@ { "label": "Overview", "to": "interrupts/overview", - "addedAt": "2026-08-04", - "updatedAt": "2026-08-16" + "addedAt": "2026-08-04" }, { "label": "Tool Approval", @@ -199,8 +198,7 @@ { "label": "Multiple Interrupts", "to": "interrupts/multiple", - "addedAt": "2026-08-04", - "updatedAt": "2026-08-16" + "addedAt": "2026-08-04" }, { "label": "Generic Interrupts", @@ -252,8 +250,7 @@ { "label": "Client Persistence", "to": "persistence/client-persistence", - "addedAt": "2026-08-04", - "updatedAt": "2026-08-16" + "addedAt": "2026-08-04" }, { "label": "Generation Persistence", @@ -740,43 +737,43 @@ "label": "@tanstack/ai-client", "to": "api/ai-client", "addedAt": "2026-04-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-08-04" }, { "label": "@tanstack/ai-react", "to": "api/ai-react", "addedAt": "2026-04-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-07-08" }, { "label": "@tanstack/ai-solid", "to": "api/ai-solid", "addedAt": "2026-04-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-07-08" }, { "label": "@tanstack/ai-preact", "to": "api/ai-preact", "addedAt": "2026-04-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-07-08" }, { "label": "@tanstack/ai-vue", "to": "api/ai-vue", "addedAt": "2026-04-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-07-30" }, { "label": "@tanstack/ai-svelte", "to": "api/ai-svelte", "addedAt": "2026-04-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-07-30" }, { "label": "@tanstack/ai-angular", "to": "api/ai-angular", "addedAt": "2026-06-15", - "updatedAt": "2026-08-16" + "updatedAt": "2026-07-30" } ] }, From 29d284932d01855034a57e14e0f9c68a936d57bc Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Thu, 20 Aug 2026 10:28:29 +0000 Subject: [PATCH 5/6] ci: apply automated fixes --- packages/ai-client/src/chat-client.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index d1a2a7ee2a..ab173aa734 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -1373,10 +1373,7 @@ export class ChatClient< resumeState, interruptState.interrupts, ) - this.callbacksRef.current.onInterruptStateChange( - interruptState, - { source }, - ) + this.callbacksRef.current.onInterruptStateChange(interruptState, { source }) } /** From b95b9936a012dfc562f5e490e14ae6950435c047 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 20 Aug 2026 15:16:17 +0200 Subject: [PATCH 6/6] fix(frameworks): keep hydrate queue open if a callback throws A throw from a queued onInterruptStateChange must still mark the client ready, or later live updates stay queued. Also set updatedAt on docs pages that describe the new hydrate source. --- docs/config.json | 23 ++++++++++++----------- packages/ai-preact/src/use-chat.ts | 20 ++++++++++++-------- packages/ai-react/src/use-chat.ts | 20 ++++++++++++-------- 3 files changed, 36 insertions(+), 27 deletions(-) diff --git a/docs/config.json b/docs/config.json index a37be59279..e853a260e3 100644 --- a/docs/config.json +++ b/docs/config.json @@ -106,7 +106,7 @@ "label": "Client Tools", "to": "tools/client-tools", "addedAt": "2026-04-15", - "updatedAt": "2026-07-21" + "updatedAt": "2026-08-20" }, { "label": "Tool Approval Flow", @@ -190,7 +190,7 @@ "label": "Overview", "to": "interrupts/overview", "addedAt": "2026-08-04", - "updatedAt": "2026-08-14" + "updatedAt": "2026-08-20" }, { "label": "Tool Approval", @@ -202,7 +202,7 @@ "label": "Multiple Interrupts", "to": "interrupts/multiple", "addedAt": "2026-08-04", - "updatedAt": "2026-08-14" + "updatedAt": "2026-08-20" }, { "label": "Generic Interrupts", @@ -276,7 +276,8 @@ { "label": "Client Persistence", "to": "persistence/client-persistence", - "addedAt": "2026-08-04" + "addedAt": "2026-08-04", + "updatedAt": "2026-08-20" }, { "label": "Generation Persistence", @@ -833,43 +834,43 @@ "label": "@tanstack/ai-client", "to": "api/ai-client", "addedAt": "2026-04-15", - "updatedAt": "2026-08-19" + "updatedAt": "2026-08-20" }, { "label": "@tanstack/ai-react", "to": "api/ai-react", "addedAt": "2026-04-15", - "updatedAt": "2026-08-19" + "updatedAt": "2026-08-20" }, { "label": "@tanstack/ai-solid", "to": "api/ai-solid", "addedAt": "2026-04-15", - "updatedAt": "2026-08-19" + "updatedAt": "2026-08-20" }, { "label": "@tanstack/ai-preact", "to": "api/ai-preact", "addedAt": "2026-04-15", - "updatedAt": "2026-08-19" + "updatedAt": "2026-08-20" }, { "label": "@tanstack/ai-vue", "to": "api/ai-vue", "addedAt": "2026-04-15", - "updatedAt": "2026-08-19" + "updatedAt": "2026-08-20" }, { "label": "@tanstack/ai-svelte", "to": "api/ai-svelte", "addedAt": "2026-04-15", - "updatedAt": "2026-08-19" + "updatedAt": "2026-08-20" }, { "label": "@tanstack/ai-angular", "to": "api/ai-angular", "addedAt": "2026-06-15", - "updatedAt": "2026-08-19" + "updatedAt": "2026-08-20" } ] }, diff --git a/packages/ai-preact/src/use-chat.ts b/packages/ai-preact/src/use-chat.ts index e2589d695a..c275d763f7 100644 --- a/packages/ai-preact/src/use-chat.ts +++ b/packages/ai-preact/src/use-chat.ts @@ -283,16 +283,20 @@ export function useChat< useEffect(() => { activeClientRef.current = client - // Keep initialization closed while draining so callbacks published by a - // queued callback are appended and delivered in the same commit. - while (initialization.callbacks.length > 0) { - if (activeClientRef.current !== client) { - initialization.callbacks.length = 0 - break + try { + // Keep initialization closed while draining so callbacks published by a + // queued callback are appended and delivered in the same commit. + while (initialization.callbacks.length > 0) { + if (activeClientRef.current !== client) { + initialization.callbacks.length = 0 + break + } + initialization.callbacks.shift()?.() } - initialization.callbacks.shift()?.() + } finally { + // A throw from a queued user callback must not leave the queue closed. + initialization.ready = true } - initialization.ready = true }, [client, initialization]) useEffect(() => { diff --git a/packages/ai-react/src/use-chat.ts b/packages/ai-react/src/use-chat.ts index 482d2fe89e..a23f321e2a 100644 --- a/packages/ai-react/src/use-chat.ts +++ b/packages/ai-react/src/use-chat.ts @@ -285,16 +285,20 @@ export function useChat< useEffect(() => { activeClientRef.current = client - // Keep initialization closed while draining so callbacks published by a - // queued callback are appended and delivered in the same commit. - while (initialization.callbacks.length > 0) { - if (activeClientRef.current !== client) { - initialization.callbacks.length = 0 - break + try { + // Keep initialization closed while draining so callbacks published by a + // queued callback are appended and delivered in the same commit. + while (initialization.callbacks.length > 0) { + if (activeClientRef.current !== client) { + initialization.callbacks.length = 0 + break + } + initialization.callbacks.shift()?.() } - initialization.callbacks.shift()?.() + } finally { + // A throw from a queued user callback must not leave the queue closed. + initialization.ready = true } - initialization.ready = true }, [client, initialization]) useEffect(() => {