fix(ai-client): preserve client-tool continuation ownership

.changeset/fix-sequential-client-tool-resumes.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-client': patch
+---
+
+Keep native interrupt ownership across sequential client-tool resumes.

docs/structured-outputs/with-tools.md

@@ -128,7 +128,7 @@ The full server-tool approval pattern lives in [Tool Approval Flow](../tools/too
 
 ## Client tools mid-run
 
-Client tools — defined with `.client((input) => ...)` on the tool definition — execute automatically when the model calls them. The runtime sees the queued `tool-input-available` custom event, looks up the registered `.client()` implementation, runs it, and posts the result back. The agent loop continues to the structured-output stream once every client tool resolves. There's no `onToolCall` option to wire up on the hook side.
+Client tools — defined with `.client((input) => ...)` on the tool definition — execute automatically when the model calls them. The server ends the current run with an internal `client-tool-execution` interrupt that does not appear in the public `interrupts` array. The client runs the registered `.client()` implementation and submits its output in a resume batch. Once every client tool resolves, the agent loop continues into the structured-output stream. There's no `onToolCall` option to wire up on the hook side.
 
 ```tsx
 import { toolDefinition } from "@tanstack/ai";

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

@@ -1852,28 +1852,17 @@ export class ChatClient<
  }
  }
 
- /**
- * True when the client still has user-actionable interrupts (or is mid
-* resume submission). Staged/submitting items that are already being
- * continued do not block a later turn once the resume stream has cleared
- * resume state.
- */
+ /** True while interrupt descriptors still own continuation. */
+private hasPendingInterrupts(): boolean {
+ return this.interruptManager.getDescriptors().length > 0
+}
+
+/** True while an interrupt batch owns the next user turn. */
  private hasBlockingInterrupts(): boolean {
- if (!this.lastResume && !this.activeInterruptSubmission) {
- return false
- }
- if (this.activeInterruptSubmission) {
- return true
- }
- return this.interruptManager
- .getInterrupts()
- .some(
- (item) =>
- item.status === 'pending' ||
- item.status === 'validating' ||
- item.status === 'error' ||
- item.status === 'staged',
- )
+ return (
+ this.activeInterruptSubmission !== undefined ||
+ this.hasPendingInterrupts()
+ )
  }
 
  /** True while a stream is active, a send is claiming the client, or the queue is draining. */
@@ -2583,6 +2572,8 @@ export class ChatClient<
  * Check if we should continue the flow and do so if needed
  */
  private async checkForContinuation(): Promise<void> {
+ if (this.hasPendingInterrupts()) return
+
  // Prevent duplicate continuation attempts
  if (this.continuationPending || this.isLoading) {
  this.continuationSkipped = true

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

@@ -1201,4 +1201,244 @@ describe('ChatClient resume', () => {
  ])
  expect(client.getInterruptState().interruptErrors).toEqual([])
  })
+
+ it('continues a legacy client tool emitted by a native resume', async () => {
+ const lookup = toolDefinition({
+ name: 'lookup',
+ description: 'Look up',
+ inputSchema: z.object({ query: z.string() }),
+ outputSchema: z.object({ answer: z.number() }),
+ }).client(async () => ({ answer: 42 }))
+ const { adapter, contexts } = recordingAdapter([
+ // The initial run pauses on a native interrupt.
+ (ctx) => [
+ {
+ type: EventType.RUN_STARTED,
+ runId: ctx?.runId ?? 'interrupted-run',
+ threadId: ctx?.threadId ?? 'thread-1',
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.RUN_FINISHED,
+ runId: ctx?.runId ?? 'interrupted-run',
+ threadId: ctx?.threadId ?? 'thread-1',
+ timestamp: Date.now(),
+ outcome: {
+ type: 'interrupt',
+ interrupts: [
+ {
+ id: 'interrupt-1',
+ reason: 'approval_required',
+ metadata: {
+ kind: 'approval',
+ toolName: 'confirm',
+ input: {},
+ },
+ },
+ ],
+ },
+ },
+ ],
+ // The native resume emits a legacy client tool, not another interrupt.
+ (ctx) => [
+ {
+ type: EventType.RUN_STARTED,
+ runId: ctx?.runId ?? 'resume-run',
+ threadId: ctx?.threadId ?? 'thread-1',
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.TOOL_CALL_START,
+ toolCallId: 'legacy-tool-call',
+ toolCallName: 'lookup',
+ toolName: 'lookup',
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.TOOL_CALL_ARGS,
+ toolCallId: 'legacy-tool-call',
+ delta: JSON.stringify({ query: 'answer' }),
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.CUSTOM,
+ name: 'tool-input-available',
+ value: {
+ toolCallId: 'legacy-tool-call',
+ toolName: 'lookup',
+ input: { query: 'answer' },
+ },
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.RUN_FINISHED,
+ runId: ctx?.runId ?? 'resume-run',
+ threadId: ctx?.threadId ?? 'thread-1',
+ finishReason: 'tool_calls',
+ timestamp: Date.now(),
+ },
+ ],
+ // The legacy tool result continues through an ordinary request.
+ (ctx) => [
+ {
+ type: EventType.RUN_STARTED,
+ runId: ctx?.runId ?? 'final-run',
+ threadId: ctx?.threadId ?? 'thread-1',
+ timestamp: Date.now(),
+ },
+ text('done'),
+ {
+ type: EventType.RUN_FINISHED,
+ runId: ctx?.runId ?? 'final-run',
+ threadId: ctx?.threadId ?? 'thread-1',
+ finishReason: 'stop',
+ timestamp: Date.now(),
+ },
+ ],
+ ])
+ const client = new ChatClient({
+ connection: adapter,
+ threadId: 'thread-1',
+ tools: [lookup],
+ })
+
+ await client.sendMessage('hi')
+ resolveGenericInterrupt(client)
+
+ await vi.waitFor(() => {
+ expect(contexts).toHaveLength(3)
+ expect(
+ client
+ .getMessages()
+ .some((message) =>
+ message.parts.some(
+ (part) => part.type === 'text' && part.content === 'done',
+ ),
+ ),
+ ).toBe(true)
+ })
+ expect(contexts[1]?.resume).toEqual([
+ {
+ interruptId: 'interrupt-1',
+ status: 'resolved',
+ payload: { answer: 'continue' },
+ },
+ ])
+ expect(contexts[2]?.resume).toBeUndefined()
+ expect(contexts[2]?.parentRunId).toBeUndefined()
+ })
+
+ it('keeps native interrupt ownership when a sequential client tool resume fails', async () => {
+ const outputSchema = z.object({ answer: z.number() })
+ const lookup = toolDefinition({
+ name: 'lookup',
+ description: 'Look up',
+ inputSchema: z.object({ query: z.string() }),
+ outputSchema,
+ }).client(async ({ query }) => ({ answer: query === 'first' ? 42 : 43 }))
+ const outputSchemaHash = hashSchemaInput(outputSchema)
+ const responseSchema = convertSchemaToJsonSchema(outputSchema) ?? {}
+ const responseSchemaHash = digestInterruptJson(
+ canonicalInterruptJson(responseSchema),
+ )
+ const interrupt =
+ (toolCallId: string, query: string): Script =>
+ (ctx) => {
+ const runId = ctx?.runId ?? `run-${toolCallId}`
+ const threadId = ctx?.threadId ?? 'thread-1'
+ return [
+ {
+ type: EventType.RUN_STARTED,
+ runId,
+ threadId,
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.TOOL_CALL_START,
+ toolCallId,
+ toolCallName: 'lookup',
+ toolName: 'lookup',
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.TOOL_CALL_ARGS,
+ toolCallId,
+ delta: JSON.stringify({ query }),
+ timestamp: Date.now(),
+ },
+ {
+ type: EventType.RUN_FINISHED,
+ runId,
+ threadId,
+ timestamp: Date.now(),
+ outcome: {
+ type: 'interrupt',
+ interrupts: [
+ {
+ id: `client_tool_${toolCallId}`,
+ reason: 'tanstack:client_tool_execution',
+ toolCallId,
+ responseSchema,
+ metadata: {
+ kind: 'client_tool',
+ toolName: 'lookup',
+ input: { query },
+ 'tanstack:interruptBinding': {
+ kind: 'client-tool-execution',
+ interruptId: `client_tool_${toolCallId}`,
+ interruptedRunId: runId,
+ generation: 0,
+ toolName: 'lookup',
+ toolCallId,
+ outputSchemaHash,
+ responseSchemaHash,
+ },
+ },
+ },
+ ],
+ },
+ },
+ ]
+ }
+
+ const { adapter, contexts } = recordingAdapter([
+ interrupt('tool-call-1', 'first'),
+ interrupt('tool-call-2', 'second'),
+ {
+ chunks: [],
+ error: new Error('resume failed'),
+ },
+ ])
+ const client = new ChatClient({
+ connection: adapter,
+ threadId: 'thread-1',
+ tools: [lookup],
+ })
+
+ await client.sendMessage('hi')
+ await vi.waitFor(() => {
+ expect(contexts).toHaveLength(3)
+ expect(client.getInterruptState().interruptErrors[0]?.code).toBe(
+ 'transport',
+ )
+ })
+
+ expect(contexts[0]?.resume).toBeUndefined()
+ expect(contexts[1]?.parentRunId).toBe(contexts[0]?.runId)
+ expect(contexts[1]?.resume).toEqual([
+ {
+ interruptId: 'client_tool_tool-call-1',
+ status: 'resolved',
+ payload: { answer: 42 },
+ },
+ ])
+ expect(contexts[2]?.parentRunId).toBe(contexts[1]?.runId)
+ expect(contexts[2]?.resume).toEqual([
+ {
+ interruptId: 'client_tool_tool-call-2',
+ status: 'resolved',
+ payload: { answer: 43 },
+ },
+ ])
+ })
 })

testing/e2e/src/routes/api.tools-test.ts

@@ -226,6 +226,8 @@ export const Route = createFileRoute('/api/tools-test')({
  context: runtimeContext,
  threadId: params.threadId,
  runId: params.runId,
+ ...(params.parentRunId ? { parentRunId: params.parentRunId } : {}),
+ ...(params.resume ? { resume: params.resume } : {}),
  agentLoopStrategy: maxIterations(20),
  abortController,
  })

testing/e2e/tests/tools-test/race-conditions.spec.ts

@@ -33,6 +33,17 @@ test.describe('Race Condition Tests', () => {
  testId,
  aimockPort,
  }) => {
+ const requestBodies: Array<any> = []
+ page.on('request', (request) => {
+ if (
+ request.url().includes('/api/tools-test') &&
+ request.method() === 'POST'
+ ) {
+ const body = request.postDataJSON()
+ if (body) requestBodies.push(body)
+ }
+ })
+
  await selectScenario(page, 'sequential-client-tools', testId, aimockPort)
 
  const startTime = Date.now()
@@ -74,6 +85,16 @@ test.describe('Race Condition Tests', () => {
  expect(executionEvents[2]?.type).toBe('execution-start')
  expect(executionEvents[3]?.type).toBe('execution-complete')
 
+ await expect(page.locator('#messages-json-content')).toContainText(
+ 'Both notifications have been shown.',
+ )
+ expect(requestBodies).toHaveLength(3)
+ expect(requestBodies[0]?.resume).toBeUndefined()
+ expect(requestBodies[1]?.parentRunId).toBe(requestBodies[0]?.runId)
+ expect(requestBodies[1]?.resume).toHaveLength(1)
+ expect(requestBodies[2]?.parentRunId).toBe(requestBodies[1]?.runId)
+ expect(requestBodies[2]?.resume).toHaveLength(1)
+
  // If it takes too long (e.g., > 10 seconds), it might indicate blocking
  // (each tool takes ~50ms, so total should be well under 5 seconds)
  expect(duration).toBeLessThan(10000)