From c73f7fdb38d556bb073273d9b3ed43140d1c06ae Mon Sep 17 00:00:00 2001 From: TPSJASPER <592827953@qq.com> Date: Sun, 25 Oct 2026 15:43:55 +0800 Subject: [PATCH 1/2] fix(ai): deliver durable MCP input answers to server tools withPersistence hands generic interrupt resolutions to the chat engine through resumeToolState. The engine copied them into resumeGenericInterrupts, but only the ephemeral path mapped mcp_input_ answers into resumeInputResponses. A durable resume therefore ran the pending server tool with ctx.inputResponse undefined, so the tool asked for input again. Apply the mapping in applyResumeToolState, which both resume paths use, and run it after the restored history and tools exist. The ephemeral path keeps the same behaviour and the old duplicate mapping goes away. Fixes #1885 --- .changeset/durable-mcp-input-response.md | 5 + docs/persistence/chat-persistence-advanced.md | 2 + .../ai-persistence/tests/interrupts.test.ts | 137 ++++++++++++++++++ packages/ai/src/activities/chat/index.ts | 33 +++-- testing/e2e/src/routeTree.gen.ts | 107 ++++++++------ .../e2e/src/routes/api.durable-mcp-input.ts | 108 ++++++++++++++ testing/e2e/tests/durable-mcp-input.spec.ts | 42 ++++++ 7 files changed, 378 insertions(+), 56 deletions(-) create mode 100644 .changeset/durable-mcp-input-response.md create mode 100644 testing/e2e/src/routes/api.durable-mcp-input.ts create mode 100644 testing/e2e/tests/durable-mcp-input.spec.ts diff --git a/.changeset/durable-mcp-input-response.md b/.changeset/durable-mcp-input-response.md new file mode 100644 index 0000000000..cd8945a171 --- /dev/null +++ b/.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. diff --git a/docs/persistence/chat-persistence-advanced.md b/docs/persistence/chat-persistence-advanced.md index 1b39f42336..dabc61ffcf 100644 --- a/docs/persistence/chat-persistence-advanced.md +++ b/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_`), 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). diff --git a/packages/ai-persistence/tests/interrupts.test.ts b/packages/ai-persistence/tests/interrupts.test.ts index ff63b08c33..d217bb8ac4 100644 --- a/packages/ai-persistence/tests/interrupts.test.ts +++ b/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 = [{ ...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 = [{ ...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()]]) diff --git a/packages/ai/src/activities/chat/index.ts b/packages/ai/src/activities/chat/index.ts index 89c7c4da44..dbc6d60896 100644 --- a/packages/ai/src/activities/chat/index.ts +++ b/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() 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) { @@ -5693,7 +5699,6 @@ class TextEngine< } private applyMiddlewareConfig(config: ChatMiddlewareConfig): void { - this.applyResumeToolState(config.resumeToolState) this.messages = config.messages if (config.activities !== undefined) { this.setActivities(config.activities) @@ -5702,6 +5707,8 @@ class TextEngine< this.systemPrompts = config.systemPrompts assertUniqueToolNames(config.tools) this.tools = config.tools + // MCP input answers need the restored history and server tool definitions. + this.applyResumeToolState(config.resumeToolState) this.params = { ...this.params, metadata: config.metadata, diff --git a/testing/e2e/src/routeTree.gen.ts b/testing/e2e/src/routeTree.gen.ts index 35cb311a76..558674e7fe 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/testing/e2e/src/routeTree.gen.ts @@ -90,6 +90,7 @@ import { Route as ApiCopilotWireRouteImport } from './routes/api.copilot-wire' import { Route as ApiCrossModelReplayWireRouteImport } from './routes/api.cross-model-replay-wire' import { Route as ApiDevtoolsMemoryRouteImport } from './routes/api.devtools-memory' import { Route as ApiDurableDeliveryRouteImport } from './routes/api.durable-delivery' +import { Route as ApiDurableMcpInputRouteImport } from './routes/api.durable-mcp-input' import { Route as ApiDurableTakeoverRouteImport } from './routes/api.durable-takeover' import { Route as ApiEmbeddingRouteImport } from './routes/api.embedding' import { Route as ApiEvaluateHttpStatusRouteImport } from './routes/api.evaluate-http-status' @@ -200,10 +201,10 @@ import { Route as ApiWebMcpPageToolsRouteImport } from './routes/api.web-mcp-pag import { Route as ApiWorldRouteImport } from './routes/api.world' import { Route as ApiWrapFetchWireRouteImport } from './routes/api.wrap-fetch-wire' import { Route as ApiAudioStreamRouteImport } from './routes/api.audio.stream' -import { Route as ApiHarnessProtocolSplatRouteImport } from './routes/api.harness-protocol.$' -import { Route as ApiHarnessSecretAnswerSplatRouteImport } from './routes/api.harness-secret-answer.$' import { Route as ApiHarnessHandoffSplatRouteImport } from './routes/api.harness-handoff.$' import { Route as ApiHarnessObservabilitySplatRouteImport } from './routes/api.harness-observability.$' +import { Route as ApiHarnessProtocolSplatRouteImport } from './routes/api.harness-protocol.$' +import { Route as ApiHarnessSecretAnswerSplatRouteImport } from './routes/api.harness-secret-answer.$' import { Route as ApiHarnessWaitsSplatRouteImport } from './routes/api.harness-waits.$' import { Route as ApiImageStreamRouteImport } from './routes/api.image.stream' import { Route as ApiTranscriptionStreamRouteImport } from './routes/api.transcription.stream' @@ -636,6 +637,11 @@ const ApiDurableDeliveryRoute = ApiDurableDeliveryRouteImport.update({ path: '/api/durable-delivery', getParentRoute: () => rootRouteImport, } as any) +const ApiDurableMcpInputRoute = ApiDurableMcpInputRouteImport.update({ + id: '/api/durable-mcp-input', + path: '/api/durable-mcp-input', + getParentRoute: () => rootRouteImport, +} as any) const ApiDurableTakeoverRoute = ApiDurableTakeoverRouteImport.update({ id: '/api/durable-takeover', path: '/api/durable-takeover', @@ -1212,17 +1218,6 @@ const ApiAudioStreamRoute = ApiAudioStreamRouteImport.update({ path: '/stream', getParentRoute: () => ApiAudioRoute, } as any) -const ApiHarnessProtocolSplatRoute = ApiHarnessProtocolSplatRouteImport.update({ - id: '/api/harness-protocol/$', - path: '/api/harness-protocol/$', - getParentRoute: () => rootRouteImport, -} as any) -const ApiHarnessSecretAnswerSplatRoute = - ApiHarnessSecretAnswerSplatRouteImport.update({ - id: '/api/harness-secret-answer/$', - path: '/api/harness-secret-answer/$', - getParentRoute: () => rootRouteImport, - } as any) const ApiHarnessHandoffSplatRoute = ApiHarnessHandoffSplatRouteImport.update({ id: '/api/harness-handoff/$', path: '/api/harness-handoff/$', @@ -1234,6 +1229,17 @@ const ApiHarnessObservabilitySplatRoute = path: '/api/harness-observability/$', getParentRoute: () => rootRouteImport, } as any) +const ApiHarnessProtocolSplatRoute = ApiHarnessProtocolSplatRouteImport.update({ + id: '/api/harness-protocol/$', + path: '/api/harness-protocol/$', + getParentRoute: () => rootRouteImport, +} as any) +const ApiHarnessSecretAnswerSplatRoute = + ApiHarnessSecretAnswerSplatRouteImport.update({ + id: '/api/harness-secret-answer/$', + path: '/api/harness-secret-answer/$', + getParentRoute: () => rootRouteImport, + } as any) const ApiHarnessWaitsSplatRoute = ApiHarnessWaitsSplatRouteImport.update({ id: '/api/harness-waits/$', path: '/api/harness-waits/$', @@ -1341,6 +1347,7 @@ export interface FileRoutesByFullPath { '/api/cross-model-replay-wire': typeof ApiCrossModelReplayWireRoute '/api/devtools-memory': typeof ApiDevtoolsMemoryRoute '/api/durable-delivery': typeof ApiDurableDeliveryRoute + '/api/durable-mcp-input': typeof ApiDurableMcpInputRoute '/api/durable-takeover': typeof ApiDurableTakeoverRoute '/api/embedding': typeof ApiEmbeddingRoute '/api/evaluate-http-status': typeof ApiEvaluateHttpStatusRoute @@ -1452,10 +1459,10 @@ export interface FileRoutesByFullPath { '/api/wrap-fetch-wire': typeof ApiWrapFetchWireRoute '/$provider/': typeof ProviderIndexRoute '/api/audio/stream': typeof ApiAudioStreamRoute - '/api/harness-protocol/$': typeof ApiHarnessProtocolSplatRoute - '/api/harness-secret-answer/$': typeof ApiHarnessSecretAnswerSplatRoute '/api/harness-handoff/$': typeof ApiHarnessHandoffSplatRoute '/api/harness-observability/$': typeof ApiHarnessObservabilitySplatRoute + '/api/harness-protocol/$': typeof ApiHarnessProtocolSplatRoute + '/api/harness-secret-answer/$': typeof ApiHarnessSecretAnswerSplatRoute '/api/harness-waits/$': typeof ApiHarnessWaitsSplatRoute '/api/image/stream': typeof ApiImageStreamRoute '/api/transcription/stream': typeof ApiTranscriptionStreamRoute @@ -1543,6 +1550,7 @@ export interface FileRoutesByTo { '/api/cross-model-replay-wire': typeof ApiCrossModelReplayWireRoute '/api/devtools-memory': typeof ApiDevtoolsMemoryRoute '/api/durable-delivery': typeof ApiDurableDeliveryRoute + '/api/durable-mcp-input': typeof ApiDurableMcpInputRoute '/api/durable-takeover': typeof ApiDurableTakeoverRoute '/api/embedding': typeof ApiEmbeddingRoute '/api/evaluate-http-status': typeof ApiEvaluateHttpStatusRoute @@ -1654,10 +1662,10 @@ export interface FileRoutesByTo { '/api/wrap-fetch-wire': typeof ApiWrapFetchWireRoute '/$provider': typeof ProviderIndexRoute '/api/audio/stream': typeof ApiAudioStreamRoute - '/api/harness-protocol/$': typeof ApiHarnessProtocolSplatRoute - '/api/harness-secret-answer/$': typeof ApiHarnessSecretAnswerSplatRoute '/api/harness-handoff/$': typeof ApiHarnessHandoffSplatRoute '/api/harness-observability/$': typeof ApiHarnessObservabilitySplatRoute + '/api/harness-protocol/$': typeof ApiHarnessProtocolSplatRoute + '/api/harness-secret-answer/$': typeof ApiHarnessSecretAnswerSplatRoute '/api/harness-waits/$': typeof ApiHarnessWaitsSplatRoute '/api/image/stream': typeof ApiImageStreamRoute '/api/transcription/stream': typeof ApiTranscriptionStreamRoute @@ -1746,6 +1754,7 @@ export interface FileRoutesById { '/api/cross-model-replay-wire': typeof ApiCrossModelReplayWireRoute '/api/devtools-memory': typeof ApiDevtoolsMemoryRoute '/api/durable-delivery': typeof ApiDurableDeliveryRoute + '/api/durable-mcp-input': typeof ApiDurableMcpInputRoute '/api/durable-takeover': typeof ApiDurableTakeoverRoute '/api/embedding': typeof ApiEmbeddingRoute '/api/evaluate-http-status': typeof ApiEvaluateHttpStatusRoute @@ -1857,10 +1866,10 @@ export interface FileRoutesById { '/api/wrap-fetch-wire': typeof ApiWrapFetchWireRoute '/$provider/': typeof ProviderIndexRoute '/api/audio/stream': typeof ApiAudioStreamRoute - '/api/harness-protocol/$': typeof ApiHarnessProtocolSplatRoute - '/api/harness-secret-answer/$': typeof ApiHarnessSecretAnswerSplatRoute '/api/harness-handoff/$': typeof ApiHarnessHandoffSplatRoute '/api/harness-observability/$': typeof ApiHarnessObservabilitySplatRoute + '/api/harness-protocol/$': typeof ApiHarnessProtocolSplatRoute + '/api/harness-secret-answer/$': typeof ApiHarnessSecretAnswerSplatRoute '/api/harness-waits/$': typeof ApiHarnessWaitsSplatRoute '/api/image/stream': typeof ApiImageStreamRoute '/api/transcription/stream': typeof ApiTranscriptionStreamRoute @@ -1950,6 +1959,7 @@ export interface FileRouteTypes { | '/api/cross-model-replay-wire' | '/api/devtools-memory' | '/api/durable-delivery' + | '/api/durable-mcp-input' | '/api/durable-takeover' | '/api/embedding' | '/api/evaluate-http-status' @@ -2061,10 +2071,10 @@ export interface FileRouteTypes { | '/api/wrap-fetch-wire' | '/$provider/' | '/api/audio/stream' - | '/api/harness-protocol/$' - | '/api/harness-secret-answer/$' | '/api/harness-handoff/$' | '/api/harness-observability/$' + | '/api/harness-protocol/$' + | '/api/harness-secret-answer/$' | '/api/harness-waits/$' | '/api/image/stream' | '/api/transcription/stream' @@ -2152,6 +2162,7 @@ export interface FileRouteTypes { | '/api/cross-model-replay-wire' | '/api/devtools-memory' | '/api/durable-delivery' + | '/api/durable-mcp-input' | '/api/durable-takeover' | '/api/embedding' | '/api/evaluate-http-status' @@ -2263,10 +2274,10 @@ export interface FileRouteTypes { | '/api/wrap-fetch-wire' | '/$provider' | '/api/audio/stream' - | '/api/harness-protocol/$' - | '/api/harness-secret-answer/$' | '/api/harness-handoff/$' | '/api/harness-observability/$' + | '/api/harness-protocol/$' + | '/api/harness-secret-answer/$' | '/api/harness-waits/$' | '/api/image/stream' | '/api/transcription/stream' @@ -2354,6 +2365,7 @@ export interface FileRouteTypes { | '/api/cross-model-replay-wire' | '/api/devtools-memory' | '/api/durable-delivery' + | '/api/durable-mcp-input' | '/api/durable-takeover' | '/api/embedding' | '/api/evaluate-http-status' @@ -2465,10 +2477,10 @@ export interface FileRouteTypes { | '/api/wrap-fetch-wire' | '/$provider/' | '/api/audio/stream' - | '/api/harness-protocol/$' - | '/api/harness-secret-answer/$' | '/api/harness-handoff/$' | '/api/harness-observability/$' + | '/api/harness-protocol/$' + | '/api/harness-secret-answer/$' | '/api/harness-waits/$' | '/api/image/stream' | '/api/transcription/stream' @@ -2557,6 +2569,7 @@ export interface RootRouteChildren { ApiCrossModelReplayWireRoute: typeof ApiCrossModelReplayWireRoute ApiDevtoolsMemoryRoute: typeof ApiDevtoolsMemoryRoute ApiDurableDeliveryRoute: typeof ApiDurableDeliveryRoute + ApiDurableMcpInputRoute: typeof ApiDurableMcpInputRoute ApiDurableTakeoverRoute: typeof ApiDurableTakeoverRoute ApiEmbeddingRoute: typeof ApiEmbeddingRoute ApiEvaluateHttpStatusRoute: typeof ApiEvaluateHttpStatusRoute @@ -2667,10 +2680,10 @@ export interface RootRouteChildren { ApiWorldRoute: typeof ApiWorldRoute ApiWrapFetchWireRoute: typeof ApiWrapFetchWireRoute ProviderIndexRoute: typeof ProviderIndexRoute - ApiHarnessProtocolSplatRoute: typeof ApiHarnessProtocolSplatRoute - ApiHarnessSecretAnswerSplatRoute: typeof ApiHarnessSecretAnswerSplatRoute ApiHarnessHandoffSplatRoute: typeof ApiHarnessHandoffSplatRoute ApiHarnessObservabilitySplatRoute: typeof ApiHarnessObservabilitySplatRoute + ApiHarnessProtocolSplatRoute: typeof ApiHarnessProtocolSplatRoute + ApiHarnessSecretAnswerSplatRoute: typeof ApiHarnessSecretAnswerSplatRoute ApiHarnessWaitsSplatRoute: typeof ApiHarnessWaitsSplatRoute } @@ -3243,6 +3256,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ApiDurableDeliveryRouteImport parentRoute: typeof rootRouteImport } + '/api/durable-mcp-input': { + id: '/api/durable-mcp-input' + path: '/api/durable-mcp-input' + fullPath: '/api/durable-mcp-input' + preLoaderRoute: typeof ApiDurableMcpInputRouteImport + parentRoute: typeof rootRouteImport + } '/api/durable-takeover': { id: '/api/durable-takeover' path: '/api/durable-takeover' @@ -4013,20 +4033,6 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ApiAudioStreamRouteImport parentRoute: typeof ApiAudioRoute } - '/api/harness-protocol/$': { - id: '/api/harness-protocol/$' - path: '/api/harness-protocol/$' - fullPath: '/api/harness-protocol/$' - preLoaderRoute: typeof ApiHarnessProtocolSplatRouteImport - parentRoute: typeof rootRouteImport - } - '/api/harness-secret-answer/$': { - id: '/api/harness-secret-answer/$' - path: '/api/harness-secret-answer/$' - fullPath: '/api/harness-secret-answer/$' - preLoaderRoute: typeof ApiHarnessSecretAnswerSplatRouteImport - parentRoute: typeof rootRouteImport - } '/api/harness-handoff/$': { id: '/api/harness-handoff/$' path: '/api/harness-handoff/$' @@ -4041,6 +4047,20 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ApiHarnessObservabilitySplatRouteImport parentRoute: typeof rootRouteImport } + '/api/harness-protocol/$': { + id: '/api/harness-protocol/$' + path: '/api/harness-protocol/$' + fullPath: '/api/harness-protocol/$' + preLoaderRoute: typeof ApiHarnessProtocolSplatRouteImport + parentRoute: typeof rootRouteImport + } + '/api/harness-secret-answer/$': { + id: '/api/harness-secret-answer/$' + path: '/api/harness-secret-answer/$' + fullPath: '/api/harness-secret-answer/$' + preLoaderRoute: typeof ApiHarnessSecretAnswerSplatRouteImport + parentRoute: typeof rootRouteImport + } '/api/harness-waits/$': { id: '/api/harness-waits/$' path: '/api/harness-waits/$' @@ -4223,6 +4243,7 @@ const rootRouteChildren: RootRouteChildren = { ApiCrossModelReplayWireRoute: ApiCrossModelReplayWireRoute, ApiDevtoolsMemoryRoute: ApiDevtoolsMemoryRoute, ApiDurableDeliveryRoute: ApiDurableDeliveryRoute, + ApiDurableMcpInputRoute: ApiDurableMcpInputRoute, ApiDurableTakeoverRoute: ApiDurableTakeoverRoute, ApiEmbeddingRoute: ApiEmbeddingRoute, ApiEvaluateHttpStatusRoute: ApiEvaluateHttpStatusRoute, @@ -4334,10 +4355,10 @@ const rootRouteChildren: RootRouteChildren = { ApiWorldRoute: ApiWorldRoute, ApiWrapFetchWireRoute: ApiWrapFetchWireRoute, ProviderIndexRoute: ProviderIndexRoute, - ApiHarnessProtocolSplatRoute: ApiHarnessProtocolSplatRoute, - ApiHarnessSecretAnswerSplatRoute: ApiHarnessSecretAnswerSplatRoute, ApiHarnessHandoffSplatRoute: ApiHarnessHandoffSplatRoute, ApiHarnessObservabilitySplatRoute: ApiHarnessObservabilitySplatRoute, + ApiHarnessProtocolSplatRoute: ApiHarnessProtocolSplatRoute, + ApiHarnessSecretAnswerSplatRoute: ApiHarnessSecretAnswerSplatRoute, ApiHarnessWaitsSplatRoute: ApiHarnessWaitsSplatRoute, } export const routeTree = rootRouteImport diff --git a/testing/e2e/src/routes/api.durable-mcp-input.ts b/testing/e2e/src/routes/api.durable-mcp-input.ts new file mode 100644 index 0000000000..7b73b827b6 --- /dev/null +++ b/testing/e2e/src/routes/api.durable-mcp-input.ts @@ -0,0 +1,108 @@ +import { createFileRoute } from '@tanstack/react-router' +import { chat } from '@tanstack/ai' +import { memoryPersistence, withPersistence } from '@tanstack/ai-persistence' +import { z } from 'zod' +import type { + AnyTextAdapter, + ToolExecutionContext, + ToolInputResponse, +} from '@tanstack/ai' + +const persistence = memoryPersistence() +const requestSchema = z.object({ + threadId: z.string(), + runId: z.string(), + parentRunId: z.string().optional(), + resume: z + .array( + z.object({ + interruptId: z.string(), + status: z.enum(['resolved', 'cancelled']), + payload: z.unknown().optional(), + }), + ) + .optional(), +}) + +export const Route = createFileRoute('/api/durable-mcp-input')({ + server: { + handlers: { + POST: async ({ request }) => { + const params = requestSchema.parse(await request.json()) + const toolCallId = `ask-city-${params.threadId}` + const responses: Array = [] + // Provider-free adapter; both requests use the real persistence middleware. + const adapter: AnyTextAdapter = { + kind: 'text', + name: 'fixed', + model: 'test-model', + '~types': {}, + async *chatStream() { + yield { + type: 'RUN_STARTED', + threadId: params.threadId, + runId: params.runId, + timestamp: 1, + } + if (!params.parentRunId) { + yield { + type: 'TOOL_CALL_START', + toolCallId, + toolCallName: 'ask_city', + timestamp: 1, + } + yield { + type: 'TOOL_CALL_ARGS', + toolCallId, + delta: '{}', + timestamp: 1, + } + } + yield { + type: 'RUN_FINISHED', + threadId: params.threadId, + runId: params.runId, + finishReason: params.parentRunId ? 'stop' : 'tool_calls', + timestamp: 1, + } + }, + structuredOutput: async () => ({ data: {}, rawText: '{}' }), + } as unknown as AnyTextAdapter + const events = [] + for await (const event of chat({ + ...params, + adapter, + messages: params.parentRunId + ? [] + : [{ role: 'user', content: 'Ask for a city' }], + tools: [ + { + name: 'ask_city', + description: 'Ask for a city', + execute: (_args: unknown, ctx?: ToolExecutionContext) => { + responses.push(ctx?.inputResponse) + if (ctx?.inputResponse) return ctx.inputResponse + throw { + name: 'MCPInputRequiredError', + kind: 'form', + request: { message: 'Which city?' }, + } + }, + }, + ], + middleware: [withPersistence(persistence)], + debug: false, + })) + events.push(event) + return Response.json({ + events, + responses, + pending: await persistence.stores.interrupts!.listPending( + params.threadId, + ), + run: await persistence.stores.runs!.get(params.runId), + }) + }, + }, + }, +}) diff --git a/testing/e2e/tests/durable-mcp-input.spec.ts b/testing/e2e/tests/durable-mcp-input.spec.ts new file mode 100644 index 0000000000..ad7f14f1b3 --- /dev/null +++ b/testing/e2e/tests/durable-mcp-input.spec.ts @@ -0,0 +1,42 @@ +import { test, expect } from './fixtures' + +for (const status of ['resolved', 'cancelled']) { + test(`durable MCP input delivers a ${status} answer across requests`, async ({ + request, + }) => { + const threadId = crypto.randomUUID() + const firstRun = crypto.randomUUID() + const first = await request.post('/api/durable-mcp-input', { + data: { threadId, runId: firstRun }, + }) + expect(first.ok()).toBe(true) + const parked = await first.json() + expect(parked.pending).toHaveLength(1) + expect(parked.run.status).toBe('interrupted') + const inputResponse = + status === 'resolved' + ? { status, payload: { value: 'Paris' } } + : { status } + const next = await request.post('/api/durable-mcp-input', { + data: { + threadId, + runId: crypto.randomUUID(), + parentRunId: firstRun, + resume: [ + { interruptId: parked.pending[0].interruptId, ...inputResponse }, + ], + }, + }) + expect(next.ok()).toBe(true) + const resumed = await next.json() + expect(resumed.responses).toEqual([inputResponse]) + expect(resumed.pending).toEqual([]) + expect(resumed.run.status).toBe('completed') + expect(resumed.events.at(-1)).toMatchObject({ type: 'RUN_FINISHED' }) + expect( + resumed.events.some( + (event: { type: string }) => event.type === 'RUN_ERROR', + ), + ).toBe(false) + }) +} From 27858f9661a9e33f4167a500b52a3e68ce25f746 Mon Sep 17 00:00:00 2001 From: Tom Beckenham <34339192+tombeckenham@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:06:53 +1100 Subject: [PATCH 2/2] fix(ai): apply resume state before middleware replaces messages Durable MCP answers arrive later from DURABLE_RESUME, after messages and tools exist. Moving this call did not deliver them. --- packages/ai/src/activities/chat/index.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/ai/src/activities/chat/index.ts b/packages/ai/src/activities/chat/index.ts index dbc6d60896..2fca84b752 100644 --- a/packages/ai/src/activities/chat/index.ts +++ b/packages/ai/src/activities/chat/index.ts @@ -5699,6 +5699,7 @@ class TextEngine< } private applyMiddlewareConfig(config: ChatMiddlewareConfig): void { + this.applyResumeToolState(config.resumeToolState) this.messages = config.messages if (config.activities !== undefined) { this.setActivities(config.activities) @@ -5707,8 +5708,6 @@ class TextEngine< this.systemPrompts = config.systemPrompts assertUniqueToolNames(config.tools) this.tools = config.tools - // MCP input answers need the restored history and server tool definitions. - this.applyResumeToolState(config.resumeToolState) this.params = { ...this.params, metadata: config.metadata,