diff --git a/.changeset/fix-1086-interrupt-hydrate-source.md b/.changeset/fix-1086-interrupt-hydrate-source.md new file mode 100644 index 000000000..48a10895f --- /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/docs/api/ai-angular.md b/docs/api/ai-angular.md index b7f22f2ad..ba0ee5288 100644 --- a/docs/api/ai-angular.md +++ b/docs/api/ai-angular.md @@ -64,6 +64,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 08b3d471e..314c88b82 100644 --- a/docs/api/ai-client.md +++ b/docs/api/ai-client.md @@ -104,6 +104,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 b121ddcda..e886b6bee 100644 --- a/docs/api/ai-preact.md +++ b/docs/api/ai-preact.md @@ -81,6 +81,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 33057ae12..35bd87f15 100644 --- a/docs/api/ai-react.md +++ b/docs/api/ai-react.md @@ -91,6 +91,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 fcb70c132..6f2ee7ac2 100644 --- a/docs/api/ai-solid.md +++ b/docs/api/ai-solid.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-svelte.md b/docs/api/ai-svelte.md index 6acd5a066..096517071 100644 --- a/docs/api/ai-svelte.md +++ b/docs/api/ai-svelte.md @@ -81,6 +81,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 832284b46..d3d476c94 100644 --- a/docs/api/ai-vue.md +++ b/docs/api/ai-vue.md @@ -79,6 +79,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 a37be5927..e853a260e 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/docs/interrupts/multiple.md b/docs/interrupts/multiple.md index c069c95be..0665f31ad 100644 --- a/docs/interrupts/multiple.md +++ b/docs/interrupts/multiple.md @@ -142,7 +142,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 resumable item with no payload. +- `cancelInterrupts()` cancels every resumable item 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 +every resumable item in the batch. ## When an answer is wrong diff --git a/docs/interrupts/overview.md b/docs/interrupts/overview.md index 09cb256c4..4708f316e 100644 --- a/docs/interrupts/overview.md +++ b/docs/interrupts/overview.md @@ -216,6 +216,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 f2d8199cb..b38673ea1 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 8f21dc04a..ffcee5c2e 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 diff --git a/packages/ai-angular/src/inject-chat.ts b/packages/ai-angular/src/inject-chat.ts index 69812f4e0..8c5869c18 100644 --- a/packages/ai-angular/src/inject-chat.ts +++ b/packages/ai-angular/src/inject-chat.ts @@ -136,9 +136,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, ...(options.interrupts !== undefined && { diff --git a/packages/ai-angular/tests/inject-chat.test.ts b/packages/ai-angular/tests/inject-chat.test.ts index 4586b1f9f..520ab3355 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 766adcd4c..750336de1 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -60,7 +60,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 { @@ -111,7 +114,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, @@ -422,6 +428,7 @@ export class ChatClient< onRunIdChange: (runId: string | null) => void onInterruptStateChange: ( state: ChatInterruptState, + context: { source: 'hydrate' | 'live' }, ) => void onCustomEvent: ( eventType: string, @@ -522,7 +529,7 @@ export class ChatClient< ? { interrupts: options.interrupts } : {}), 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 @@ -882,7 +889,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 @@ -890,16 +897,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', + ) } /** @@ -1128,12 +1138,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 } @@ -1180,7 +1193,7 @@ export class ChatClient< this.interruptManager.reset() return } - this.notifyResumeStateChange() + this.notifyResumeStateChange('live') } /** @@ -1400,19 +1413,20 @@ 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(), - ) - this.callbacksRef.current.onInterruptStateChange( - this.interruptManager.getState(), + interruptState.interrupts, ) + this.callbacksRef.current.onInterruptStateChange(interruptState, { source }) } /** diff --git a/packages/ai-client/src/interrupt-manager.ts b/packages/ai-client/src/interrupt-manager.ts index 90144c35c..5d4c67566 100644 --- a/packages/ai-client/src/interrupt-manager.ts +++ b/packages/ai-client/src/interrupt-manager.ts @@ -50,6 +50,8 @@ export interface InterruptManagerSubmission { fingerprint: string } +export type InterruptManagerChangeSource = 'hydrate' | 'live' + export interface InterruptManagerOptions< TTools extends ReadonlyArray, TInterrupts extends ReadonlyArray> = @@ -58,7 +60,7 @@ export interface InterruptManagerOptions< tools?: TTools interrupts?: TInterrupts submit: (submission: InterruptManagerSubmission) => Promise - onChange?: () => void + onChange?: (source: InterruptManagerChangeSource) => void } type UnknownObject = { [key: string]: unknown } @@ -538,7 +540,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, @@ -567,7 +572,7 @@ export class InterruptManager< this.submissionRootErrors = Object.freeze([]) this.retrySubmission = undefined this.resuming = false - this.publish() + this.publish(source) } getInterrupts(): BoundInterrupts { @@ -588,7 +593,10 @@ export class InterruptManager< ) } - reset(options?: { preserveRootErrors?: boolean }): void { + reset(options?: { + preserveRootErrors?: boolean + source?: InterruptManagerChangeSource + }): void { this.hydration = undefined this.items = [] this.snapshot = Object.freeze([]) @@ -604,7 +612,7 @@ export class InterruptManager< interruptErrors: this.rootErrors, resuming: false, }) - this.options.onChange?.() + this.options.onChange?.(options?.source ?? 'live') } getInterruptErrors(): ReadonlyArray { @@ -1057,7 +1065,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({ @@ -1066,7 +1076,7 @@ export class InterruptManager< interruptErrors: this.rootErrors, resuming: this.resuming, }) - this.options.onChange?.() + this.options.onChange?.(source) return } this.snapshot = this.buildSnapshot() @@ -1076,7 +1086,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 510b6c343..60ca9e4d6 100644 --- a/packages/ai-client/src/types.ts +++ b/packages/ai-client/src/types.ts @@ -943,9 +943,14 @@ export interface ChatClientBaseOptions< */ onRunIdChange?: (runId: string | null) => void - /** Callback when the immutable interrupt state snapshot changes. */ + /** + * 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 /** diff --git a/packages/ai-client/tests/chat-client-interrupts.test.ts b/packages/ai-client/tests/chat-client-interrupts.test.ts index 5fe406d22..f3f280873 100644 --- a/packages/ai-client/tests/chat-client-interrupts.test.ts +++ b/packages/ai-client/tests/chat-client-interrupts.test.ts @@ -1323,11 +1323,84 @@ 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('reports source hydrate when a first-party generic interrupt is restored', () => { + const onInterruptStateChange = vi.fn() + const approval = defineInterrupt({ + id: 'approval', + responseSchema: z.object({ answer: z.number() }), + }) + const binding: InterruptBinding = { + v: INTERRUPT_BINDING_VERSION, + kind: 'generic', + interruptId: 'generic_review_one', + interruptedRunId: 'run-1', + generation: 0, + definitionId: 'approval', + key: 'one', + batchIndex: 0, + responseSchemaHash: digestInterruptJson( + canonicalInterruptJson( + convertSchemaToJsonSchema(approval.responseSchema), + ), + ), + } + const client = new ChatClient({ + connection: { async *connect() {} }, + interrupts: [approval], + onInterruptStateChange, + initialResumeSnapshot: { + resumeState: { threadId: 'thread-1', runId: 'run-1' }, + pendingInterrupts: [descriptor(binding)], + }, + }) + + expect(client.getInterrupts()[0]?.kind).toBe('generic') + 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> = [] @@ -1376,9 +1449,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 edf3d9e87..4a4e19104 100644 --- a/packages/ai-client/tests/interrupts-types.test-d.ts +++ b/packages/ai-client/tests/interrupts-types.test-d.ts @@ -160,8 +160,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 e4e41feaf..bb108b44b 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' @@ -369,6 +375,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 e98cc99ba..c275d763f 100644 --- a/packages/ai-preact/src/use-chat.ts +++ b/packages/ai-preact/src/use-chat.ts @@ -99,7 +99,7 @@ export function useChat< messagesRef.current = messages }, [messages]) - const client = useMemo(() => { + const { client, initialization } = useMemo(() => { const messagesToUse = options.initialMessages || [] isFirstMountRef.current = false @@ -122,7 +122,24 @@ export function useChat< } return currentInstance } - const pendingInitializationErrors: Array = [] + // 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) => { + if (!initializationState.ready) { + initializationState.callbacks.push(callback) + return + } + const currentInstance = instanceHolder.current + if (!currentInstance || activeClientRef.current !== currentInstance) + return + callback() + } const instance = new ChatClient({ devtoolsBridgeFactory: createChatDevtoolsBridge, ...transport, @@ -162,25 +179,24 @@ 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) => { - 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 - optionsRef.current.onCustomEvent?.(eventType, data, context) + runOrQueueForActiveInstance(() => { + optionsRef.current.onCustomEvent?.(eventType, data, context) + }) }, ...(initialOptions.tools !== undefined && { tools: initialOptions.tools, @@ -192,69 +208,97 @@ 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) => { - 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, initialization: initializationState } }, [clientId, syncResumeState]) + useEffect(() => { + activeClientRef.current = client + 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()?.() + } + } finally { + // A throw from a queued user callback must not leave the queue closed. + initialization.ready = true + } + }, [client, initialization]) + useEffect(() => { const clientMessages = client.getMessages() if (clientMessages !== messagesRef.current) { @@ -321,7 +365,6 @@ export function useChat< clearTimeout(cleanupInvalidationRef.current) cleanupInvalidationRef.current = null } - activeClientRef.current = client client.mountDevtools() // Delivery-durability resume is transparent: the resumable SSE connection // adapter reattaches via the browser's native Last-Event-ID on reconnect. diff --git a/packages/ai-preact/tests/use-chat.test.ts b/packages/ai-preact/tests/use-chat.test.ts index 149a59224..222f3fb91 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,9 +74,47 @@ describe('useChat', () => { interrupts: result.current.interrupts, interruptErrors: result.current.interruptErrors, }), + { source: 'live' }, ) }) + 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) { + if (interrupt.kind === 'unbound') continue + 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') @@ -180,6 +223,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 = [ @@ -1067,25 +1136,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 4b4034f33..a23f321e2 100644 --- a/packages/ai-react/src/use-chat.ts +++ b/packages/ai-react/src/use-chat.ts @@ -103,7 +103,7 @@ export function useChat< }, []) // Create ChatClient instance with callbacks to sync state - const client = useMemo(() => { + const { client, initialization } = useMemo(() => { const messagesToUse = options.initialMessages || [] isFirstMountRef.current = false @@ -126,7 +126,24 @@ export function useChat< } return currentInstance } - const pendingInitializationErrors: Array = [] + // 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) => { + if (!initializationState.ready) { + initializationState.callbacks.push(callback) + return + } + const currentInstance = instanceHolder.current + if (!currentInstance || activeClientRef.current !== currentInstance) + return + callback() + } const instance = new ChatClient({ devtoolsBridgeFactory: createChatDevtoolsBridge, ...transport, @@ -159,25 +176,24 @@ export function useChat< outputKind: initialOptions.outputSchema ? 'structured' : 'chat', }, onResponse: (response) => { + // ChatClient awaits this return value. Queuing would drop the promise. if (!getActiveInstance()) return - void optionsRef.current.onResponse?.(response) + return 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) => { - 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, @@ -186,76 +202,105 @@ export function useChat< interrupts: initialOptions.interrupts, }), 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) => { - 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, initialization: initializationState } }, [clientId, syncResumeState]) + useEffect(() => { + activeClientRef.current = client + 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()?.() + } + } finally { + // A throw from a queued user callback must not leave the queue closed. + initialization.ready = true + } + }, [client, initialization]) + useEffect(() => { const clientMessages = client.getMessages() if (clientMessages !== messagesRef.current) { @@ -339,7 +384,6 @@ export function useChat< clearTimeout(cleanupInvalidationRef.current) cleanupInvalidationRef.current = null } - activeClientRef.current = client client.mountDevtools() // Delivery-durability resume is transparent: the resumable SSE connection // adapter re-attaches via the browser's native Last-Event-ID on reconnect. diff --git a/packages/ai-react/tests/use-chat.test.ts b/packages/ai-react/tests/use-chat.test.ts index a72d247a5..c2665c89a 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 { @@ -37,12 +37,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({ @@ -74,9 +83,47 @@ describe('useChat', () => { interrupts: result.current.interrupts, interruptErrors: result.current.interruptErrors, }), + { source: 'live' }, ) }) + 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) { + if (interrupt.kind === 'unbound') continue + 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') @@ -185,6 +232,79 @@ 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('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 = [ diff --git a/packages/ai-solid/src/use-chat.ts b/packages/ai-solid/src/use-chat.ts index eb11d87a8..41ce0fdb8 100644 --- a/packages/ai-solid/src/use-chat.ts +++ b/packages/ai-solid/src/use-chat.ts @@ -193,9 +193,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 fe0fcd48b..492ae85e6 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 dfa9e4f45..cf3a01f39 100644 --- a/packages/ai-svelte/src/create-chat.svelte.ts +++ b/packages/ai-svelte/src/create-chat.svelte.ts @@ -190,9 +190,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 e9ee4b732..297826ef3 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 522c27ce9..e31d8d9ad 100644 --- a/packages/ai-vue/src/use-chat.ts +++ b/packages/ai-vue/src/use-chat.ts @@ -176,9 +176,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 6e92f4b25..65a019ee7 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 8f7aae0a7..6394f1c6e 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 c050047a6..42d7618a4 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')