fix(ai): pass durable MCP input answers to pending server tools

.changeset/durable-mcp-input-response.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai': patch
+---
+
+Pass resolved and cancelled MCP input responses to pending server tools when resuming through persistence with an empty incoming message history.

docs/persistence/chat-persistence-advanced.md

@@ -249,6 +249,8 @@ Persistence is the **server-authoritative resume path**. For each request with a
 3. It **clears** `config.resume`, so the chat engine skips its own short-lived rebuild. That rebuild needs client message history, and the persistence flow does not send it.
 4. It commits all resolved and cancelled entries at one success boundary: the next interrupt or the finish.
 
+For an MCP input interrupt (`mcp_input_<toolCallId>`), the pending server tool runs again with the answer in `ctx.inputResponse`. A resolved answer carries `{ status: 'resolved', payload }`. A cancelled answer carries `{ status: 'cancelled' }`. This also works with `messages: []` when the message store restores the thread.
+
 An `InterruptStore` can implement `commitBatch()` to make that commit atomic. Without it, the fallback writes the entries one at a time, and the write is not atomic.
 
 `onInterruptResolution` runs at the same moment as in a request without persistence: after the init `onConfig`, before `onStart`. See [Apply Answers](../interrupts/apply-answers).

packages/ai-persistence/tests/interrupts.test.ts

@@ -11,6 +11,7 @@ import type {
   StreamChunk,
   TokenUsage,
   Tool,
+  ToolExecutionContext,
 } from '@tanstack/ai'
 import { memoryPersistence } from '../src/memory'
 import { withPersistence } from '../src/middleware'
@@ -228,6 +229,142 @@ const approvalClientTool = (name: string): Tool => ({
 })
 
 describe('interrupt persistence', () => {
+  it.each([
+    { statuses: ['resolved'] },
+    { statuses: ['cancelled'] },
+    { statuses: ['resolved', 'resolved', 'cancelled'] },
+  ] as const)(
+    'passes durable MCP input responses $statuses to their pending server tools',
+    async ({ statuses }) => {
+      const persistence = memoryPersistence()
+      const callIds = statuses.map((_, index) => `tool-call-${index + 1}`)
+      const execute = vi.fn((_args: unknown, ctx?: ToolExecutionContext) => {
+        if (ctx?.inputResponse) return ctx.inputResponse
+        throw {
+          name: 'MCPInputRequiredError',
+          kind: 'form',
+          request: { message: 'Which city?' },
+        }
+      })
+      const tools: Array<Tool> = [{ ...clientTool('clientSearch'), execute }]
+      const first = await collect(
+        chat({
+          adapter: mockAdapter([
+            [
+              runStarted(),
+              ...callIds.flatMap((toolCallId) => [
+                { ...toolStart(), toolCallId },
+                { ...toolArgs(), toolCallId },
+              ]),
+              toolCallFinished(),
+            ],
+          ]).adapter,
+          messages: [{ role: 'user', content: 'Ask for a city' }],
+          tools,
+          threadId: 't1',
+          runId: 'r1',
+          middleware: [withPersistence(persistence)],
+          debug: false,
+        }),
+      )
+      expect(first.at(-1)).toMatchObject({
+        type: EventType.RUN_FINISHED,
+        outcome: {
+          type: 'interrupt',
+          interrupts: callIds.map((id) => ({ id: `mcp_input_${id}` })),
+        },
+      })
+      expect(
+        await persistence.stores.interrupts!.listPending('t1'),
+      ).toHaveLength(callIds.length)
+      const responses = statuses.map((status, index) =>
+        status === 'resolved'
+          ? { status, payload: { value: index === 0 ? 'Paris' : 'London' } }
+          : { status },
+      )
+      const resumed = mockAdapter([
+        [runStarted(), text('complete'), runFinished('r2')],
+      ])
+      const chunks = await collect(
+        chat({
+          adapter: resumed.adapter,
+          messages: [],
+          tools,
+          threadId: 't1',
+          runId: 'r2',
+          parentRunId: 'r1',
+          resume: responses.map((response, index) => ({
+            interruptId: `mcp_input_${callIds[index]}`,
+            ...response,
+          })),
+          middleware: [withPersistence(persistence)],
+          debug: false,
+        }),
+      )
+
+      expect(execute).toHaveBeenCalledTimes(callIds.length * 2)
+      for (const [index, toolCallId] of callIds.entries()) {
+        const resumedCall = execute.mock.calls
+          .slice(callIds.length)
+          .find((call) => call[1]?.toolCallId === toolCallId)
+        expect(resumedCall?.[1]?.inputResponse).toEqual(responses[index])
+        expect(
+          (await persistence.stores.interrupts!.get(`mcp_input_${toolCallId}`))
+            ?.status,
+        ).toBe(statuses[index])
+      }
+      expect(chunks.some((chunk) => chunk.type === EventType.RUN_ERROR)).toBe(
+        false,
+      )
+      expect(chunks.some(isInterruptTerminal)).toBe(false)
+      expect(chunks.at(-1)).toMatchObject({ type: EventType.RUN_FINISHED })
+      expect((await persistence.stores.runs!.get('r2'))?.status).toBe(
+        'completed',
+      )
+      expect(resumed.calls).toHaveLength(1)
+      expect(await persistence.stores.interrupts!.listPending('t1')).toEqual([])
+    },
+  )
+
+  it('rejects a durable MCP answer for a different tool call', async () => {
+    const persistence = memoryPersistence()
+    const execute = vi.fn(() => {
+      throw {
+        name: 'MCPInputRequiredError',
+        kind: 'form',
+        request: { message: 'Which city?' },
+      }
+    })
+    const tools: Array<Tool> = [{ ...clientTool('clientSearch'), execute }]
+    await persistClientToolTurn(persistence, tools)
+    const resumed = mockAdapter([[runStarted(), runFinished('r2')]])
+    const chunks = await collect(
+      chat({
+        adapter: resumed.adapter,
+        threadId: 't1',
+        runId: 'r2',
+        parentRunId: 'r1',
+        messages: [],
+        tools,
+        resume: [
+          {
+            interruptId: 'mcp_input_unknown',
+            status: 'resolved',
+            payload: 'Paris',
+          },
+        ],
+        middleware: [withPersistence(persistence)],
+        debug: false,
+      }),
+    )
+    expectResumeError(chunks, 'mcp_input_tool-call-1')
+    expect(execute).toHaveBeenCalledTimes(1)
+    expect(resumed.calls).toHaveLength(0)
+    expect(await persistence.stores.interrupts!.listPending('t1')).toHaveLength(
+      1,
+    )
+  })
+
   it('persists RUN_FINISHED interrupt outcomes as pending interrupt records', async () => {
     const persistence = memoryPersistence()
     const { adapter } = mockAdapter([[runStarted(), interruptFinished()]])

packages/ai/src/activities/chat/index.ts

@@ -5374,12 +5374,10 @@ class TextEngine<
     pending.push(...genericPending)
     // An `mcp_input_*` answer is valid only for a server tool call that is
     // still pending in history.
-    const mcpInputCallIds = new Map<string, string>()
     for (const toolCall of pendingToolCalls) {
       const interruptId = `mcp_input_${toolCall.id}`
       if (!resumeInterruptIds.has(interruptId)) continue
       if (!toolsByCallId.get(toolCall.id)?.execute) continue
-      mcpInputCallIds.set(interruptId, toolCall.id)
       pending.push({
         interruptId,
         payload: { id: interruptId },
@@ -5421,16 +5419,6 @@ class TextEngine<
     })
 
     const genericResolutions = validated.resumeToolState.genericInterrupts
-    for (const [interruptId, toolCallId] of mcpInputCallIds) {
-      const resolution = genericResolutions?.get(interruptId)
-      if (!resolution) continue
-      this.resumeInputResponses.set(
-        toolCallId,
-        resolution.status === 'resolved'
-          ? { status: 'resolved', payload: resolution.payload }
-          : { status: 'cancelled' },
-      )
-    }
     if (genericPending.length > 0 && genericResolutions) {
       const resolutions = genericPending
         .sort((left, right) => {
@@ -5641,6 +5629,24 @@ class TextEngine<
       for (const [interruptId, resolution] of state.genericInterrupts) {
         this.resumeGenericInterrupts.set(interruptId, resolution)
       }
+      for (const toolCall of this.getPendingToolCallsFromMessages()) {
+        if (
+          !this.tools.some(
+            (tool) => tool.name === toolCall.function.name && tool.execute,
+          )
+        )
+          continue
+        const resolution = state.genericInterrupts.get(
+          `mcp_input_${toolCall.id}`,
+        )
+        if (!resolution) continue
+        this.resumeInputResponses.set(
+          toolCall.id,
+          resolution.status === 'resolved'
+            ? { status: 'resolved', payload: resolution.payload }
+            : { status: 'cancelled' },
+        )
+      }
     }
     if (state?.genericInterruptRequests) {
       for (const [interruptId, request] of state.genericInterruptRequests) {