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 when this call starts the stream. If a stream is already in progress, this call queues the send and the returned promise can resolve before that queued 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,20 @@ 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
+  const extra = chunk as StreamChunk & { finishReason?: unknown }
+  if (extra.finishReason !== undefined) {
+    return extra.finishReason === 'tool_calls'
+  }
+  return tanstackMetadata(chunk)?.finishReason === 'tool_calls'
+}
+
 export interface NormalizedQueueConfig {
   whenBusy: WhenBusy
   drain: 'fifo' | 'batch'
@@ -416,6 +430,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 +538,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 +1162,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 +2368,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 +3077,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,28 @@ export function normalizeConnectionAdapter(
     }
   }
 
+  async function waitUntilSubscriberIdle(
+    abortSignal?: AbortSignal,
+  ): Promise<void> {
+    // Idle means the subscriber is waiting for the next chunk, so the
+    // previous chunk has left processIncomingChunk. Empty waiters with an
+    // empty buffer is in-flight delivery, not idle.
+    const idle = () =>
+      activeBuffer.length === 0 &&
+      (activeWaiters.length > 0 || abortSignal?.aborted)
+    for (let i = 0; i < 16 && !abortSignal?.aborted; i++) {
+      if (idle()) return
+      await Promise.resolve()
+    }
+    let macrotaskWaits = 0
+    while (!abortSignal?.aborted) {
+      if (idle()) 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 +1184,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,113 @@ 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()
+    })
+
+    it('keeps append pending when intermediate tool_calls is a direct finishReason', 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(),
+            finishReason: 'tool_calls',
+          }
+          await Promise.resolve()
+          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-direct', 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-direct' }),
+      ])
+      expect(client.getResumeState()?.runId).toBeTruthy()
+    })
   })
 
   describe('reload', () => {

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

@@ -212,6 +212,7 @@ export class StreamProcessor {
   private finishReason: string | null = null
   private hasError = false
   private isDone = false
+  private streamEndEmitted = false
 
   // Recording
   private recording: ChunkRecording | null = null
@@ -729,9 +730,26 @@ 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
   }
 
+  private resumeAssistantState(id: string, state: MessageStreamState): void {
+    this.activeMessageIds.add(id)
+    if (state.isComplete || this.isDone) {
+      state.isComplete = false
+      this.isDone = false
+    }
+  }
+
   /**
    * Ensure an active assistant message exists, creating one if needed.
    * Used for backward compat when events arrive without prior TEXT_MESSAGE_START.
@@ -748,14 +766,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.resumeAssistantState(preferredId, state)
+        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.resumeAssistantState(activeId, state)
+        return { messageId: activeId, state }
+      }
     }
 
     // Check if a message with preferredId already exists (reconnect/resume case).
@@ -1647,8 +1671,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 +2374,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
@@ -2407,7 +2438,8 @@ export class StreamProcessor {
     }
 
     // Emit stream end for the last assistant message
-    if (lastAssistantMessage) {
+    if (lastAssistantMessage && !this.streamEndEmitted) {
+      this.streamEndEmitted = true
       this.events.onStreamEnd?.(lastAssistantMessage)
     }
   }
@@ -2526,6 +2558,7 @@ export class StreamProcessor {
     this.finishReason = null
     this.hasError = false
     this.isDone = false
+    this.streamEndEmitted = false
     this.chunkStrategy.reset?.()
   }