fix(ai-client): do not end append on an intermediate tool_calls RUN_FINISHED

.changeset/quiet-append-drain.md

@@ -0,0 +1,6 @@
+---
+'@tanstack/ai': patch
+'@tanstack/ai-client': patch
+---
+
+Keep `append()` pending until the HTTP response is fully processed, including later `RUN_FINISHED` events in the same agent loop.

docs/api/ai-client.md

@@ -176,6 +176,8 @@ export async function POST(request: Request) {
 
 Appends a message to the conversation. If you pass a `UIMessage`, `append` copies `uiMessage.metadata` onto the stored message.
 
+`append()` resolves after the full HTTP response is processed. A `RUN_FINISHED` with `finishReason: "tool_calls"` does not end the wait when the agent loop continues in that response.
+
 ```typescript
 import { client } from "./client";
 import type { UIMessage } from "@tanstack/ai-client";

packages/ai-client/src/chat-client.ts

@@ -164,6 +164,16 @@ function resolveTransport(transport: {
   throw new Error('ChatClient: either `connection` or `fetcher` is required.')
 }
 
+function connectionDrainsOnSend(connection: ConnectionAdapter): boolean {
+  return 'connect' in connection
+}
+
+function isIntermediateToolTurn(chunk: StreamChunk): boolean {
+  if (chunk.type !== 'RUN_FINISHED') return false
+  if (chunk.outcome?.type === 'interrupt') return false
+  return tanstackMetadata(chunk)?.finishReason === 'tool_calls'
+}
+
 export interface NormalizedQueueConfig {
   whenBusy: WhenBusy
   drain: 'fifo' | 'batch'
@@ -416,6 +426,12 @@ export class ChatClient<
   private continuationPending = false
   private subscriptionAbortController: AbortController | null = null
   private processingResolve: (() => void) | null = null
+  /**
+   * `connect()` adapters push the full HTTP body into the subscribe queue, then
+   * wait until that queue is idle. After `send()` returns, every chunk from this
+   * request has been processed. Subscribe/send sockets do not drain that way.
+   */
+  private connectionDrainsOnSend = false
   private errorReportedGeneration: number | null = null
   private streamGeneration = 0
   private continuationGeneration = 0
@@ -518,7 +534,9 @@ export class ChatClient<
     this.byokProvider = options.byokProvider
     this.context = options.context
     this.queueConfig = normalizeQueueOption(options.queue)
-    this.connection = normalizeConnectionAdapter(resolveTransport(options))
+    const transport = resolveTransport(options)
+    this.connectionDrainsOnSend = connectionDrainsOnSend(transport)
+    this.connection = normalizeConnectionAdapter(transport)
 
     // Build client tools map
     this.clientToolsRef = { current: new Map() }
@@ -1140,7 +1158,9 @@ export class ChatClient<
       this.clearedStreamTracker.onSessionRunError()
     }
     this.setSessionGenerating(this.activeRunIds.size > 0)
-    if (options?.resolveProcessing !== false) {
+    const skipProcessingResolve =
+      chunk.type === 'RUN_FINISHED' && isIntermediateToolTurn(chunk)
+    if (options?.resolveProcessing !== false && !skipProcessingResolve) {
       this.resolveProcessing()
     }
   }
@@ -2344,6 +2364,14 @@ export class ChatClient<
         return false
       }
 
+      // connect() send() already waited until the subscribe queue was idle.
+      // Kick the processing wait so a stream that ends on tool_calls (no
+      // interrupt / stop) cannot hang. Subscribe/send sockets still wait for
+      // a request-ending terminal below.
+      if (this.connectionDrainsOnSend) {
+        this.resolveProcessing()
+      }
+
       // Wait for subscription loop to finish processing all chunks
       await processingComplete
 
@@ -3045,12 +3073,12 @@ export class ChatClient<
       this.resetSessionGenerating()
       this.setIsSubscribed(false)
       this.setConnectionStatus('disconnected')
-      this.connection = normalizeConnectionAdapter(
-        resolveTransport({
-          connection: options.connection,
-          fetcher: options.fetcher,
-        }),
-      )
+      const transport = resolveTransport({
+        connection: options.connection,
+        fetcher: options.fetcher,
+      })
+      this.connectionDrainsOnSend = connectionDrainsOnSend(transport)
+      this.connection = normalizeConnectionAdapter(transport)
 
       if (wasSubscribed) {
         this.subscribe()

packages/ai-client/src/connection-adapters.ts

@@ -1055,6 +1055,27 @@ export function normalizeConnectionAdapter(
     }
   }
 
+  async function waitUntilSubscriberIdle(
+    abortSignal?: AbortSignal,
+  ): Promise<void> {
+    const idle = () =>
+      activeBuffer.length === 0 &&
+      (activeWaiters.length > 0 || abortSignal?.aborted)
+    for (let i = 0; i < 16 && !abortSignal?.aborted; i++) {
+      if (idle()) return
+      if (activeBuffer.length === 0 && activeWaiters.length === 0) return
+      await Promise.resolve()
+    }
+    let macrotaskWaits = 0
+    while (!abortSignal?.aborted) {
+      if (idle()) return
+      if (activeBuffer.length === 0 && activeWaiters.length === 0) return
+      await new Promise<void>((resolve) => setTimeout(resolve, 0))
+      macrotaskWaits++
+      if (activeWaiters.length === 0 && macrotaskWaits >= 32) return
+    }
+  }
+
   return {
     subscribe(abortSignal?: AbortSignal): AsyncIterable<StreamChunk> {
       // Transfer ownership to the latest subscriber so only one active
@@ -1162,6 +1183,7 @@ export function normalizeConnectionAdapter(
         }
         throw err
       }
+      await waitUntilSubscriberIdle(abortSignal)
     },
     // Expose joinRun only when the underlying connection is resumable. Require
     // a real function — `'joinRun' in connection` is true for

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

@@ -2306,6 +2306,58 @@ describe('ChatClient', () => {
       expect(messages[0]?.id).toBeTruthy()
       expect(messages[0]?.createdAt).toBeInstanceOf(Date)
     })
+
+    it('keeps append pending through an intermediate tool_calls RUN_FINISHED until the interrupt', async () => {
+      const adapter: ConnectConnectionAdapter = {
+        async *connect(_messages, _data, _signal, ctx) {
+          const runId = ctx?.runId ?? 'run-1'
+          const threadId = ctx?.threadId ?? 'thread-1'
+          yield {
+            type: EventType.RUN_STARTED,
+            runId,
+            threadId,
+            timestamp: Date.now(),
+          }
+          yield {
+            type: EventType.RUN_FINISHED,
+            runId,
+            threadId,
+            timestamp: Date.now(),
+            metadata: { tanstack: { finishReason: 'tool_calls' } },
+          }
+          yield {
+            type: EventType.RUN_STARTED,
+            runId: 'provider-2',
+            threadId,
+            timestamp: Date.now(),
+          }
+          yield {
+            type: EventType.RUN_FINISHED,
+            runId: 'provider-2',
+            threadId,
+            timestamp: Date.now(),
+            outcome: {
+              type: 'interrupt',
+              interrupts: [{ id: 'interrupt-1', reason: 'client_tool_input' }],
+            },
+          }
+        },
+      }
+      const client = new ChatClient({
+        connection: adapter,
+        threadId: 'thread-1',
+      })
+
+      await client.append({
+        role: 'user',
+        content: 'Notify me',
+      })
+
+      expect(client.getPendingInterrupts()).toEqual([
+        expect.objectContaining({ id: 'interrupt-1' }),
+      ])
+      expect(client.getResumeState()?.runId).toBeTruthy()
+    })
   })
 
   describe('reload', () => {

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

@@ -729,6 +729,15 @@ export class StreamProcessor {
         return id
       }
     }
+    // finalizeStream() clears activeMessageIds but keeps messageStates.
+    // Leftover reasoning after an early RUN_FINISHED must resume that
+    // assistant. A new user turn calls prepareAssistantMessage(), which
+    // clears messageStates first.
+    for (const [id, state] of [...this.messageStates].reverse()) {
+      if (state.role === 'assistant') {
+        return id
+      }
+    }
     return null
   }
 
@@ -748,14 +757,20 @@ export class StreamProcessor {
     // Try to find state by preferred ID
     if (preferredId) {
       const state = this.getMessageState(preferredId)
-      if (state) return { messageId: preferredId, state }
+      if (state) {
+        this.activeMessageIds.add(preferredId)
+        return { messageId: preferredId, state }
+      }
     }
 
     // Try active assistant message
     const activeId = this.getActiveAssistantMessageId()
     if (activeId) {
       const state = this.getMessageState(activeId)
-      if (state) return { messageId: activeId, state }
+      if (state) {
+        this.activeMessageIds.add(activeId)
+        return { messageId: activeId, state }
+      }
     }
 
     // Check if a message with preferredId already exists (reconnect/resume case).
@@ -1647,8 +1662,14 @@ export class StreamProcessor {
     }
 
     if (this.activeRuns.size === 0) {
-      this.isDone = true
       this.completeAllToolCalls()
+      const isIntermediateToolTurn =
+        this.finishReason === 'tool_calls' &&
+        chunk.outcome?.type !== 'interrupt'
+      if (isIntermediateToolTurn) {
+        return
+      }
+      this.isDone = true
       this.finalizeStream()
     }
   }
@@ -2344,6 +2365,7 @@ export class StreamProcessor {
    * @see docs/chat-architecture.md#single-shot-text-response — Finalization step
    */
   finalizeStream(): void {
+    this.isDone = true
     let lastAssistantMessage: UIMessage | undefined
 
     // Finalize ALL active messages

packages/ai/tests/stream-processor.test.ts

@@ -2976,7 +2976,9 @@ describe('StreamProcessor', () => {
       expect(state.toolCalls.size).toBe(1)
       expect(state.toolCallOrder).toEqual(['tc-1'])
       expect(state.finishReason).toBe('tool_calls')
-      expect(state.done).toBe(true)
+      expect(state.done).toBe(false)
+      processor.finalizeStream()
+      expect(processor.getState().done).toBe(true)
     })
 
     it('should return independent copies (mutations do not affect internal state)', () => {
@@ -4603,6 +4605,55 @@ describe('StreamProcessor', () => {
       expect(processor.getState().done).toBe(true)
     })
 
+    it('appends leftover reasoning after a stop RUN_FINISHED onto the same thinking part', () => {
+      const processor = new StreamProcessor()
+      processor.prepareAssistantMessage()
+
+      processor.processChunk(ev.runStarted())
+      processor.processChunk(ev.stepStarted('step-1'))
+      processor.processChunk(
+        ev.reasoningContent(
+          'The user is asking for a beginner guitar recommen.',
+        ),
+      )
+      processor.processChunk(ev.runFinished('stop'))
+      processor.processChunk(ev.reasoningContent(' $1,299.'))
+
+      const messages = processor.getMessages()
+      expect(messages).toHaveLength(1)
+      const thinkingParts = messages[0]!.parts.filter(
+        (part) => part.type === 'thinking',
+      )
+      expect(thinkingParts).toHaveLength(1)
+      const thinkingPart = thinkingParts[0]
+      if (thinkingPart?.type !== 'thinking') {
+        throw new Error('expected a thinking part')
+      }
+      expect(thinkingPart.content).toBe(
+        'The user is asking for a beginner guitar recommen. $1,299.',
+      )
+    })
+
+    it('does not fire onStreamEnd on a sequential tool_calls terminal', () => {
+      const events = spyEvents()
+      const processor = new StreamProcessor({ events })
+
+      processor.processChunk(ev.runStarted('run-1'))
+      processor.processChunk(ev.textStart('msg-1'))
+      processor.processChunk(ev.textContent('calling', 'msg-1'))
+      processor.processChunk(ev.runFinished('tool_calls', 'run-1'))
+
+      expect(events.onStreamEnd).not.toHaveBeenCalled()
+      expect(processor.getState().done).toBe(false)
+
+      processor.processChunk(ev.runStarted('run-2'))
+      processor.processChunk(ev.textContent(' done', 'msg-1'))
+      processor.processChunk(ev.runFinished('stop', 'run-2'))
+
+      expect(events.onStreamEnd).toHaveBeenCalledTimes(1)
+      expect(processor.getState().done).toBe(true)
+    })
+
     it('single run should finalize normally (backward compat)', () => {
       const events = spyEvents()
       const processor = new StreamProcessor({ events })
@@ -4656,6 +4707,8 @@ describe('StreamProcessor', () => {
       expect(processor.getState().toolCalls.get('tc-a')?.state).toBe(
         'input-complete',
       )
+      expect(processor.getState().done).toBe(false)
+      processor.finalizeStream()
       expect(processor.getState().done).toBe(true)
     })
 

testing/e2e/src/lib/tools-test-tools.ts

@@ -253,6 +253,11 @@ export const SCENARIO_LIST = [
     label: 'Client Tool Input Error',
     category: 'basic',
   },
+  {
+    id: 'invalid-client-tool-retry',
+    label: 'Invalid Client Tool Retry (Regression #1192)',
+    category: 'race',
+  },
   // Race condition / event flow scenarios
   {
     id: 'sequential-client-tools',
@@ -312,6 +317,7 @@ export function getToolsForScenario(scenario: string) {
     case 'client-tool-reasoning':
     case 'client-tool-stop':
     case 'client-tool-input-error':
+    case 'invalid-client-tool-retry':
       return [clientToolDefinitions.show_notification]
 
     case 'server-context':