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..2fca84b752 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) { 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) + }) +}