fix(ai-client): drain client-tool results after joinRun replay

.changeset/join-run-drain-client-tool.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-client': patch
+---
+
+Drain queued post-stream actions after a `joinRun` replay so a client-tool result from a rejoin is sent back to the server.

docs/config.json

@@ -243,7 +243,7 @@
           "label": "Advanced",
           "to": "resumable-streams/advanced",
           "addedAt": "2026-08-04",
-          "updatedAt": "2026-08-19"
+          "updatedAt": "2026-08-20"
         },
         {
           "label": "WebSockets",

docs/resumable-streams/advanced.md

@@ -101,7 +101,9 @@ A durable run's producer is decoupled from the delivery socket. When the client
 disconnects (a page reload, a dropped connection), the response is cancelled but
 the run keeps draining into the log to its own terminal, so a reconnect or a
 mount-time `joinRun` tails it to completion. This is what makes a run resumable
-across a full reload, not just an in-session reconnect.
+across a full reload, not just an in-session reconnect. If that replay ends on
+a client tool, the client sends the tool result after the replay finishes. That
+is the same path as a live stream.
 
 A run ends early only on a genuine cancel or a failure:
 

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

@@ -1662,6 +1662,12 @@ export class ChatClient<
           }
           await this.processIncomingChunk(chunk, { defer: false })
         }
+        // Same contract as `streamResponse`: client tools may finish (and
+        // queue a resume) while `isLoading` is still true. Wait for them
+        // before teardown so `drainPostStreamActions` below sees the queue.
+        if (this.pendingToolExecutions.size > 0) {
+          await Promise.all(this.pendingToolExecutions.values())
+        }
       } catch (error) {
         // Pre-attach failures (unknown/evicted run, connect deadline abort)
         // stay soft: keep the restored transcript. Post-attach transport/parser
@@ -1703,6 +1709,7 @@ export class ChatClient<
           this.abortController = null
           this.setIsLoading(false)
           if (this.status === 'streaming') this.setStatus('ready')
+          await this.drainPostStreamActions()
         }
       }
     })()

packages/ai-client/tests/chat-client-join-run-client-tool.test.ts

@@ -0,0 +1,211 @@
+import { describe, expect, it, vi } from 'vitest'
+import {
+  EventType,
+  canonicalInterruptJson,
+  convertSchemaToJsonSchema,
+  digestInterruptJson,
+  hashSchemaInput,
+  toolDefinition,
+} from '@tanstack/ai/client'
+import { z } from 'zod'
+import { ChatClient } from '../src/chat-client'
+import { createUIMessage } from './test-utils'
+import type {
+  ResumableConnectConnectionAdapter,
+  RunAgentInputContext,
+} from '../src/connection-adapters'
+import type { StreamChunk } from '@tanstack/ai/client'
+import type { ChatClientPersistence, ChatPersistedState } from '../src/types'
+
+function memoryPersistence(initial: ChatPersistedState): ChatClientPersistence {
+  let value: ChatPersistedState | undefined = initial
+  return {
+    getItem: () => value,
+    setItem: (_id, state) => {
+      value = state
+    },
+    removeItem: () => {
+      value = undefined
+    },
+  }
+}
+
+function mountedChatClient(
+  options: ConstructorParameters<typeof ChatClient>[0],
+) {
+  const client = new ChatClient(options)
+  client.attach()
+  return client
+}
+
+function createDeferred() {
+  let resolve!: () => void
+  const promise = new Promise<void>((res) => {
+    resolve = res
+  })
+  return { promise, resolve }
+}
+
+/**
+ * Regression for https://github.com/TanStack/ai/issues/1058
+ *
+ * A joinRun replay that ends on a client tool queues the resume while
+ * `isLoading` is true. Live `streamResponse` drains that queue in `finally`.
+ * Rejoin used its own teardown and never drained, so the tool result never
+ * went back to the server and later sends stayed blocked.
+ */
+describe('joinRun client-tool continuation (issue #1058)', () => {
+  it('posts the client-tool result after joinRun replay ends', async () => {
+    const outputSchema = z.object({ answer: z.number() })
+    const toolGate = createDeferred()
+    const lookup = toolDefinition({
+      name: 'lookup',
+      description: 'Look up',
+      inputSchema: z.object({ query: z.string() }),
+      outputSchema,
+    }).client(async () => {
+      await toolGate.promise
+      return { answer: 42 }
+    })
+    const outputSchemaHash = hashSchemaInput(outputSchema)
+    const responseSchema = convertSchemaToJsonSchema(outputSchema) ?? {}
+    const responseSchemaHash = digestInterruptJson(
+      canonicalInterruptJson(responseSchema),
+    )
+
+    const joinHold = createDeferred()
+    const connectContexts: Array<RunAgentInputContext | undefined> = []
+    let connectCount = 0
+
+    const replayChunks: Array<StreamChunk> = [
+      {
+        type: EventType.RUN_STARTED,
+        runId: 'r1',
+        threadId: 't1',
+        timestamp: 1,
+      },
+      {
+        type: EventType.TOOL_CALL_START,
+        toolCallId: 'tool-call-1',
+        toolCallName: 'lookup',
+        toolName: 'lookup',
+        timestamp: 2,
+      },
+      {
+        type: EventType.TOOL_CALL_ARGS,
+        toolCallId: 'tool-call-1',
+        delta: '{"query":"first"}',
+        timestamp: 3,
+      },
+      {
+        type: EventType.RUN_FINISHED,
+        runId: 'r1',
+        threadId: 't1',
+        timestamp: 4,
+        outcome: {
+          type: 'interrupt',
+          interrupts: [
+            {
+              id: 'client_tool_tool-call-1',
+              reason: 'tanstack:client_tool_execution',
+              toolCallId: 'tool-call-1',
+              responseSchema,
+              metadata: {
+                kind: 'client_tool',
+                toolName: 'lookup',
+                input: { query: 'first' },
+                'tanstack:interruptBinding': {
+                  kind: 'client-tool-execution',
+                  interruptId: 'client_tool_tool-call-1',
+                  interruptedRunId: 'r1',
+                  generation: 0,
+                  toolName: 'lookup',
+                  toolCallId: 'tool-call-1',
+                  outputSchemaHash,
+                  responseSchemaHash,
+                },
+              },
+            },
+          ],
+        },
+      },
+    ]
+
+    const joinRun = vi.fn(async function* (_runId: string) {
+      for (const chunk of replayChunks) {
+        yield chunk
+      }
+      await joinHold.promise
+    })
+
+    const connection: ResumableConnectConnectionAdapter = {
+      async *connect(_messages, _data, _signal, runContext) {
+        connectCount++
+        connectContexts.push(runContext)
+        yield {
+          type: EventType.RUN_STARTED,
+          runId: runContext?.runId ?? 'r2',
+          threadId: runContext?.threadId ?? 't1',
+          timestamp: 10,
+        }
+        yield {
+          type: EventType.TEXT_MESSAGE_CONTENT,
+          messageId: 'm1',
+          timestamp: 11,
+          delta: 'done',
+        }
+        yield {
+          type: EventType.RUN_FINISHED,
+          runId: runContext?.runId ?? 'r2',
+          threadId: runContext?.threadId ?? 't1',
+          timestamp: 12,
+          finishReason: 'stop',
+        }
+      },
+      joinRun,
+    }
+
+    const client = mountedChatClient({
+      threadId: 't1',
+      connection,
+      persistence: memoryPersistence({
+        messages: [createUIMessage('user-1', 'hi', 'user')],
+        resume: {
+          resumeState: { threadId: 't1', runId: 'r1' },
+        },
+      }),
+      tools: [lookup],
+    })
+
+    await vi.waitFor(() => {
+      expect(joinRun).toHaveBeenCalledWith('r1', expect.anything())
+    })
+    for (let i = 0; i < 10; i++) {
+      await new Promise((r) => setTimeout(r, 0))
+    }
+
+    toolGate.resolve()
+    for (let i = 0; i < 10; i++) {
+      await new Promise((r) => setTimeout(r, 0))
+    }
+
+    joinHold.resolve()
+
+    await vi.waitFor(() => {
+      expect(connectCount).toBe(1)
+    })
+
+    expect(connectContexts[0]?.parentRunId).toBe('r1')
+    expect(connectContexts[0]?.resume).toEqual([
+      {
+        interruptId: 'client_tool_tool-call-1',
+        status: 'resolved',
+        payload: { answer: 42 },
+      },
+    ])
+    expect(client.getInterruptState().interruptErrors).toEqual([])
+
+    await client.sendMessage('later')
+    expect(connectCount).toBe(2)
+  })
+})

testing/e2e/README.md

@@ -92,6 +92,7 @@ Notes:
 | -------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `tests/delivery-durability.spec.ts`    | Transport layer: offset-tagged log, `Last-Event-ID` reconnect, second-tab join, SSE + NDJSON                                                                |
 | `tests/persistence-durability.spec.ts` | Client layer: a browser refresh restores the conversation and any pending interrupt                                                                         |
+| `tests/join-run-client-tool.spec.ts`   | Mid-run reload: `joinRun` replay that ends on a client tool drains the continuation (issue #1058)                                                           |
 | `tests/sandbox-durability.spec.ts`     | Sandbox instances: a second run resumes the persisted sandbox                                                                                               |
 | `tests/durable-takeover.spec.ts`       | Takeover with log alignment, detach-on-disconnect, out-of-band cancel in both bands, cancel-vs-disconnect divergence, and the superseded-driver epoch fence |
 
@@ -127,6 +128,7 @@ E2E coverage is mandatory for every feature, bug fix, or behavior change (see th
 | Tool system change                      | Add scenario to `tools-test-scenarios.ts` + test in `tools-test/` specs.                                                                                                                                   |
 | Middleware change                       | Add test to `middleware.spec.ts` with appropriate scenario.                                                                                                                                                |
 | Client-side change (useChat, etc.)      | Add test covering the observable behavior change.                                                                                                                                                          |
+| joinRun + client-tool continuation      | Add a spec that reloads mid-run, tails with `joinRun`, and asserts the client-tool result POSTs after replay (see `tests/join-run-client-tool.spec.ts`).                                                   |
 | Durable/detachable run + takeover       | Add a spec that disconnects mid-stream, reconnects with the same `runId`, and asserts the resumed stream continues from the aligned offset instead of restarting the agent.                                |
 | Out-of-band cancel vs. plain disconnect | Add a case distinguishing `requestRunCancel`/`wasCancelRequested` (durable cancel — record ends `'aborted'`) from an unrequested disconnect on a detachable run (record stays `'running'`, resumable).     |
 | New run-store backend                   | Run `runPersistenceConformance` from `packages/ai-persistence/src/testkit/` against it; it now pins `undefined` vs. explicit `false` on `cancelRequested` and absent vs. explicit `undefined` on `update`. |