diff --git a/packages/mcp/package.json b/packages/mcp/package.json index e563df2da1..95eaa10d9f 100644 --- a/packages/mcp/package.json +++ b/packages/mcp/package.json @@ -7,7 +7,8 @@ "private": true, "exports": { ".": "./dist/index.js", - "./test-only/stdio-server": "./dist/__fixtures__/stdio-server.js" + "./test-only/stdio-server": "./dist/__fixtures__/stdio-server.js", + "./test-only/form-server": "./dist/__fixtures__/form-server.js" }, "main": "./dist/index.js", "types": "./dist/index.d.ts", diff --git a/packages/mcp/src/__fixtures__/form-server.ts b/packages/mcp/src/__fixtures__/form-server.ts new file mode 100644 index 0000000000..2779ef045b --- /dev/null +++ b/packages/mcp/src/__fixtures__/form-server.ts @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { createServer, type IncomingHttpHeaders } from 'node:http'; +import type { CallToolResult, ElicitResult, Tool } from '@modelcontextprotocol/client'; +import { toNodeHandler } from '@modelcontextprotocol/node'; +import { createMcpHandler, inputRequired, Server } from '@modelcontextprotocol/server'; + +export interface McpFormWireCall { + id: string | number; + params: { + name: string; + arguments?: Record; + _meta?: Record; + inputResponses?: Record; + requestState?: string; + }; +} + +type FormServerResult = CallToolResult | ReturnType; + +export interface McpFormFixture { + url: string; + calls: McpFormWireCall[]; + callHeaders: IncomingHttpHeaders[]; + initializeCapabilities?: unknown; + definition: Tool; + respond(params: McpFormWireCall['params']): FormServerResult | Promise; + close(): Promise; +} + +/** Real modern-only server, with wire capture before the SDK dispatches it. */ +export async function createMcpFormFixture( + options: { legacy?: boolean } = {}, +): Promise { + const errors: unknown[] = []; + const handler = createMcpHandler( + () => { + const server = new Server( + { name: 'form-fixture', version: '1' }, + { + capabilities: { tools: {} }, + }, + ); + server.setRequestHandler('tools/list', async () => ({ tools: [fixture.definition] })); + server.setRequestHandler('tools/call', async ({ params }, context) => + fixture.respond({ + ...params, + ...(context.mcpReq.inputResponses === undefined + ? {} + : { inputResponses: context.mcpReq.inputResponses as Record }), + ...(context.mcpReq.requestState() === undefined + ? {} + : { requestState: context.mcpReq.requestState() }), + }), + ); + return server; + }, + { + legacy: options.legacy ? 'stateless' : 'reject', + keepAliveMs: 0, + onerror: (error) => errors.push(error), + }, + ); + const handle = toNodeHandler(handler, { onerror: (error) => errors.push(error) }); + const server = createServer(async (req, res) => { + try { + const chunks: Buffer[] = []; + for await (const chunk of req) chunks.push(Buffer.from(chunk)); + const text = Buffer.concat(chunks).toString('utf8'); + const body = text ? JSON.parse(text) : undefined; + for (const message of Array.isArray(body) ? body : [body]) { + if (message?.method === 'initialize') + fixture.initializeCapabilities = message.params?.capabilities; + if (message?.method === 'tools/call') { + fixture.calls.push(structuredClone(message)); + fixture.callHeaders.push({ ...req.headers }); + } + } + await handle(req, res, body); + } catch (error) { + errors.push(error); + if (!res.headersSent) res.writeHead(500); + res.end(); + } + }); + const fixture: McpFormFixture = { + url: '', + calls: [], + callHeaders: [], + definition: { name: 'ask_user', inputSchema: { type: 'object' } }, + respond: (params) => + params.inputResponses + ? { content: [{ type: 'text', text: 'complete' }] } + : inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: 'opaque-state', + }), + close: async () => { + await handler.close(); + await new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ); + if (errors.length) throw new AggregateError(errors, 'MCP form fixture failed'); + }, + }; + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const address = server.address(); + if (!address || typeof address === 'string') throw new Error('MCP form fixture has no address'); + fixture.url = `http://127.0.0.1:${address.port}/mcp`; + return fixture; +} + +export function mcpFixtureFormRequest() { + return { + method: 'elicitation/create' as const, + params: { + mode: 'form' as const, + message: 'Please confirm your details', + requestedSchema: { + type: 'object' as const, + properties: { + name: { type: 'string' as const }, + email: { type: 'string' as const, format: 'email' as const }, + confirm: { type: 'boolean' as const }, + }, + required: ['name', 'email', 'confirm'], + }, + }, + }; +} diff --git a/packages/mcp/src/__fixtures__/form-stdio-server.ts b/packages/mcp/src/__fixtures__/form-stdio-server.ts new file mode 100644 index 0000000000..e3d29a0f0d --- /dev/null +++ b/packages/mcp/src/__fixtures__/form-stdio-server.ts @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { setTimeout as delay } from 'node:timers/promises'; +import { Server, inputRequired } from '@modelcontextprotocol/server'; +import { serveStdio } from '@modelcontextprotocol/server/stdio'; +import { mcpFixtureFormRequest } from './form-server.js'; + +const STATE = 'stdio-private-continuation-state'; +const FUTURE_STATE = `future-state-prefix-${'x'.repeat(3_000)}`; +serveStdio( + () => { + const server = new Server( + { name: 'modern-stdio-form', version: '1' }, + { + capabilities: { tools: {} }, + }, + ); + server.setRequestHandler('tools/list', async () => ({ + tools: [ + { + name: 'ask_user', + inputSchema: { type: 'object' }, + }, + ], + })); + let generationCall = 0; + let lastState = STATE; + let firstState = ''; + server.setRequestHandler('tools/call', async (request, context) => { + const mode = request.params.arguments?.mode; + if (mode === 'future-diagnostic') { + process.stderr.write(`${FUTURE_STATE}\n`); + await delay(30); + return { content: [{ type: 'text', text: 'diagnosed' }] }; + } + if (mode === 'unicode-diagnostic') { + const encoded = Buffer.from(lastState); + process.stderr.write(encoded.subarray(0, 1)); + await delay(30); + process.stderr.write(encoded.subarray(1)); + process.stderr.write('\nsafe diagnostic\n'); + await delay(30); + return { content: [{ type: 'text', text: 'diagnosed' }] }; + } + if (mode === 'first-diagnostic') { + process.stderr.write(`earliest continuation: ${firstState}\n`); + await delay(30); + return { content: [{ type: 'text', text: 'diagnosed' }] }; + } + if (mode === 'diagnostic') { + process.stderr.write(`late continuation: ${lastState}\nsafe diagnostic\n`); + await delay(30); + return { content: [{ type: 'text', text: 'diagnosed' }] }; + } + const answer = context.mcpReq.inputResponses?.form; + if (answer === undefined) { + generationCall += 1; + lastState = + mode === 'future' + ? FUTURE_STATE + : mode === 'unicode' + ? `敏感-${STATE}` + : mode === 'control' + ? `${STATE}\ncontrol-private-suffix` + : mode === 'retention' + ? `${STATE}-${generationCall}` + : mode === 'byte-retention' + ? `${STATE}-${generationCall}-${'x'.repeat(16_000)}` + : STATE; + firstState ||= lastState; + if (mode === 'before') { + process.stderr.write(`before response: ${lastState}\n`); + await delay(30); + } + return inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: lastState, + }); + } + assert.equal(context.mcpReq.requestState(), lastState); + const completedState = lastState; + if (mode === 'partial' || mode === 'continued') { + process.stderr.write(lastState.slice(0, 12) + (mode === 'continued' ? '\\\n' : '')); + await delay(30); + setTimeout( + () => process.stderr.write(`${completedState.slice(12)}\nsafe diagnostic\n`), + 60, + ); + } else { + process.stderr.write(`continuation: ${lastState}\n`); + await delay(30); + setTimeout( + () => process.stderr.write(`after completion: ${completedState}\nsafe diagnostic\n`), + 60, + ); + } + return { content: [{ type: 'text', text: 'Form completed' }], structuredContent: { answer } }; + }); + return server; + }, + { legacy: 'reject' }, +); diff --git a/packages/mcp/src/__tests__/form-elicitation.test.ts b/packages/mcp/src/__tests__/form-elicitation.test.ts new file mode 100644 index 0000000000..8773931c3d --- /dev/null +++ b/packages/mcp/src/__tests__/form-elicitation.test.ts @@ -0,0 +1,314 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import type { InteractionFormResult } from '@maka/core/interaction'; +import { prepareMcpForm } from '../form-elicitation.js'; + +const requester = { name: 'ask_user', source: 'Example server' }; +function request(properties: Record, required: string[] = []) { + return { + method: 'elicitation/create', + params: { + message: 'Please provide details', + requestedSchema: { type: 'object', properties, required }, + }, + }; +} + +function invalidAnswer(prepared: ReturnType, result: unknown) { + assert.throws( + () => prepared.respond(result as InteractionFormResult), + /^Error: Invalid MCP form response$/, + ); +} + +describe('MCP form elicitation adapter', () => { + test('preserves types, defaults, optional fields and validation constraints', () => { + const prepared = prepareMcpForm( + request( + { + name: { type: 'string', title: 'Your name', minLength: 2, maxLength: 10, default: 'Ada' }, + email: { type: 'string', format: 'email', maxLength: 100, description: 'Contact email' }, + age: { type: 'integer', minimum: 18, maximum: 120, default: 30 }, + amount: { type: 'number', minimum: 0, maximum: 1, default: 0.5 }, + confirm: { type: 'boolean', default: false }, + }, + ['name', 'confirm'], + ), + requester, + ); + assert.deepEqual(prepared.form.requester, requester); + assert.deepEqual( + prepared.form.fields.map((field) => field.kind), + ['string', 'string', 'integer', 'number', 'boolean'], + ); + assert.deepEqual(prepared.form.fields[0], { + kind: 'string', + name: 'name', + label: 'Your name', + required: true, + minLength: 2, + maxLength: 10, + default: 'Ada', + }); + assert.deepEqual( + prepared.respond({ action: 'accept', values: { name: 'Ada', confirm: false } }), + { action: 'accept', content: { name: 'Ada', confirm: false } }, + ); + invalidAnswer(prepared, { action: 'accept', values: { name: 'A', confirm: true } }); + invalidAnswer(prepared, { action: 'accept', values: { name: 'Ada' } }); + invalidAnswer(prepared, { + action: 'accept', + values: { name: 'Ada', confirm: true, age: 18.5 }, + }); + invalidAnswer(prepared, { + action: 'accept', + values: { name: 'Ada', confirm: true, email: 'invalid' }, + }); + invalidAnswer(prepared, { + action: 'accept', + values: { name: 'Ada', confirm: true, extra: 'unrequested' }, + }); + }); + + test('applies a visible client limit to unbounded server strings', () => { + const prepared = prepareMcpForm( + request( + { + name: { type: 'string' }, + email: { type: 'string', format: 'email' }, + confirm: { type: 'boolean' }, + }, + ['name', 'email', 'confirm'], + ), + requester, + ); + assert.equal((prepared.form.fields[0] as { maxLength: number }).maxLength, 256); + assert.equal((prepared.form.fields[1] as { maxLength: number }).maxLength, 256); + assert.deepEqual( + prepared.respond({ + action: 'accept', + values: { name: 'Ada', email: 'ada@example.com', confirm: true }, + }), + { action: 'accept', content: { name: 'Ada', email: 'ada@example.com', confirm: true } }, + ); + invalidAnswer(prepared, { + action: 'accept', + values: { name: 'x'.repeat(257), email: 'ada@example.com', confirm: true }, + }); + for (const schema of [ + { type: 'string', minLength: 257 }, + { type: 'string', default: 'x'.repeat(257) }, + ]) { + assert.throws(() => prepareMcpForm(request({ field: schema }), requester)); + } + assert.throws(() => + prepareMcpForm( + request( + Object.fromEntries( + Array.from({ length: 6 }, (_, index) => [String(index), { type: 'string' }]), + ), + ), + requester, + ), + ); + const explicit = prepareMcpForm( + request({ field: { type: 'string', maxLength: 300 } }), + requester, + ); + assert.equal((explicit.form.fields[0] as { maxLength: number }).maxLength, 300); + }); + + test('normalizes all protocol enum variants without changing choice values', () => { + const prepared = prepareMcpForm( + request({ + plain: { type: 'string', enum: ['a', 'b'], default: 'a' }, + legacy: { type: 'string', enum: ['a', 'b'], enumNames: ['Alpha', 'Beta'] }, + titled: { + type: 'string', + oneOf: [ + { const: 'a', title: 'Alpha' }, + { const: 'b', title: 'Beta' }, + ], + }, + multi: { + type: 'array', + items: { type: 'string', enum: ['a', 'b'] }, + minItems: 1, + maxItems: 2, + default: ['a'], + }, + titledMulti: { + type: 'array', + items: { + anyOf: [ + { const: 'a', title: 'Alpha' }, + { const: 'b', title: 'Beta' }, + ], + }, + }, + }), + requester, + ); + assert.deepEqual( + prepared.form.fields.map((field) => field.kind), + ['single_select', 'single_select', 'single_select', 'multi_select', 'multi_select'], + ); + for (const index of [1, 2, 4]) + assert.deepEqual((prepared.form.fields[index] as { options: unknown }).options, [ + { value: 'a', label: 'Alpha' }, + { value: 'b', label: 'Beta' }, + ]); + assert.deepEqual( + prepared.respond({ + action: 'accept', + values: { legacy: 'a', multi: ['b'], titledMulti: [] }, + }), + { action: 'accept', content: { legacy: 'a', multi: ['b'], titledMulti: [] } }, + ); + invalidAnswer(prepared, { action: 'accept', values: { plain: 'c' } }); + invalidAnswer(prepared, { action: 'accept', values: { multi: [] } }); + invalidAnswer(prepared, { action: 'accept', values: { multi: ['a', 'a'] } }); + }); + + test('returns bare decline and cancel responses and rejects attached answers', () => { + const prepared = prepareMcpForm( + request({ confirm: { type: 'boolean' } }, ['confirm']), + requester, + ); + assert.deepEqual(prepared.respond({ action: 'decline' }), { action: 'decline' }); + assert.deepEqual(prepared.respond({ action: 'cancel' }), { action: 'cancel' }); + invalidAnswer(prepared, { action: 'decline', values: { confirm: true } }); + invalidAnswer(prepared, { action: 'accept', values: { confirm: 'true' } }); + }); + + test('rejects unsupported constraints and impossible schemas before showing a form', () => { + for (const schema of [ + { type: 'object', properties: {} }, + { type: 'string', pattern: 'secret-pattern' }, + { type: 'number', exclusiveMinimum: 0 }, + { type: 'string', enum: ['a'], minLength: 2 }, + { type: 'string', enum: ['a'], enumNames: ['A', 'B'] }, + { type: 'string', oneOf: [{ const: 'a', title: 'A', pattern: 'x' }] }, + { type: 'array', items: { type: 'string' } }, + { type: 'array', items: { type: 'string', enum: ['a'] }, minItems: 2 }, + { type: 'string', minLength: 2, maxLength: 1 }, + { type: 'integer', minimum: 0.1, maximum: 0.9 }, + { type: 'number', minimum: 2, default: 1 }, + { type: 'string', title: null }, + { type: 'string', enum: ['a', 'a'] }, + { + type: 'string', + oneOf: [ + { const: 'a', title: 'same' }, + { const: 'b', title: 'same' }, + ], + }, + { type: 'boolean', default: 'false' }, + ]) + assert.throws( + () => prepareMcpForm(request({ field: schema }), requester), + /^Error: Unsupported or invalid MCP form request$/, + ); + assert.throws(() => + prepareMcpForm(request({ field: { type: 'string' } }, ['missing']), requester), + ); + assert.throws(() => + prepareMcpForm(request({ field: { type: 'string' } }, ['field', 'field']), requester), + ); + }); + + test('rejects non-form requests and unsupported root constraints', () => { + const original = request({ field: { type: 'string' } }); + assert.throws(() => + prepareMcpForm( + { + ...original, + params: { + ...original.params, + requestedSchema: { ...original.params.requestedSchema, required: null }, + }, + }, + requester, + ), + ); + for (const input of [ + null, + [], + { ...original, method: 'sampling/createMessage' }, + { ...original, params: { ...original.params, mode: 'url' } }, + { + ...original, + params: { + ...original.params, + requestedSchema: { + ...original.params.requestedSchema, + dependentRequired: { field: ['other'] }, + }, + }, + }, + ]) { + assert.throws( + () => prepareMcpForm(input, requester), + /^Error: Unsupported or invalid MCP form request$/, + ); + } + }); + + test('uses canonical form bounds and strips display controls without changing protocol values', () => { + const prepared = prepareMcpForm( + request({ + field: { + type: 'string', + title: 'Your\u001b[31m name', + maxLength: 100, + default: 'literal\nvalue', + }, + }), + requester, + ); + assert.equal(prepared.form.fields[0]?.label.includes('\u001b'), false); + assert.equal(prepared.form.fields[0]?.default, undefined); + assert.deepEqual(prepared.respond({ action: 'accept', values: { field: 'literal\nvalue' } }), { + action: 'accept', + content: { field: 'literal\nvalue' }, + }); + assert.throws(() => + prepareMcpForm( + request({ field: { type: 'string', default: 'x'.repeat(100_000) } }), + requester, + ), + ); + const bounded = prepareMcpForm( + request({ field: { type: 'string', maxLength: 100 } }), + requester, + ); + invalidAnswer(bounded, { action: 'accept', values: { field: 'x'.repeat(100_000) } }); + }); + + test('fails closed when canonical admission cannot represent a field identity', () => { + const properties = JSON.parse('{"__proto__":{"type":"string","maxLength":100}}'); + assert.throws( + () => prepareMcpForm(request(properties, ['__proto__']), requester), + /^Error: Unsupported or invalid MCP form request$/, + ); + }); +}); diff --git a/packages/mcp/src/__tests__/form-manager.test.ts b/packages/mcp/src/__tests__/form-manager.test.ts new file mode 100644 index 0000000000..22c469a046 --- /dev/null +++ b/packages/mcp/src/__tests__/form-manager.test.ts @@ -0,0 +1,704 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { afterEach, test } from 'node:test'; +import { setTimeout as delay } from 'node:timers/promises'; +import { CLIENT_CAPABILITIES_META_KEY, ProtocolError } from '@modelcontextprotocol/client'; +import { inputRequired } from '@modelcontextprotocol/server'; +import { MCP_CONFIG_VERSION } from '@maka/core/mcp'; +import type { InteractionFormResult } from '@maka/core/interaction'; +import { McpClientManager } from '../index.js'; +import { createMcpFormFixture, mcpFixtureFormRequest } from '../__fixtures__/form-server.js'; + +const resources: Array<() => Promise> = []; +afterEach(async () => { + for (const close of resources.splice(0).reverse()) await close(); +}); + +const accepted: InteractionFormResult = { + action: 'accept', + values: { name: 'Ada', email: 'ada@example.com', confirm: true }, +}; + +async function setup(legacy = false) { + const fixture = await createMcpFormFixture({ legacy }); + resources.push(() => fixture.close()); + const manager = new McpClientManager({ timeouts: { callToolMs: 1_000 } }); + resources.push(() => manager.close()); + await manager.sync({ + version: MCP_CONFIG_VERSION, + mcpServers: { + forms: { + transport: 'streamable-http', + url: fixture.url, + protocol: legacy ? 'legacy' : 'auto', + }, + }, + }); + const binding = manager.toolSnapshot().tools[0]!.binding; + return { fixture, manager, binding }; +} + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason: unknown) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; +} + +test('real modern form round trip preserves original arguments, IDs, state and per-call capability', async () => { + const { fixture, manager, binding } = await setup(); + const args = { source: { value: 'original' } }; + const result = await manager.callTool(binding, args, { + requestInteraction: async (form) => { + assert.deepEqual(form.requester, { name: 'ask_user', source: 'forms' }); + assert.deepEqual( + form.fields.map((field) => field.name), + ['name', 'email', 'confirm'], + ); + assert.equal(JSON.stringify(form).includes('opaque-state'), false); + args.source.value = 'changed'; + return accepted; + }, + }); + assert.equal(result.content[0]?.type, 'text'); + assert.equal(fixture.calls.length, 2); + assert.notEqual(fixture.calls[0]!.id, fixture.calls[1]!.id); + for (const call of fixture.calls) { + assert.deepEqual(call.params.arguments, { source: { value: 'original' } }); + assert.deepEqual(call.params._meta?.[CLIENT_CAPABILITIES_META_KEY], { + elicitation: { form: {} }, + }); + } + assert.equal(fixture.calls[1]!.params.requestState, 'opaque-state'); + assert.deepEqual(fixture.calls[1]!.params.inputResponses, { + form: { action: 'accept', content: accepted.values }, + }); +}); + +for (const action of ['decline', 'cancel'] as const) { + test(`explicit ${action} is a protocol response, not global Stop`, async () => { + const { fixture, manager, binding } = await setup(); + await manager.callTool(binding, {}, { requestInteraction: async () => ({ action }) }); + assert.deepEqual(fixture.calls[1]!.params.inputResponses, { form: { action } }); + }); +} + +test('handler-less call does not advertise elicitation and does not retry', async () => { + const { fixture, manager, binding } = await setup(); + await assert.rejects(manager.callTool(binding, {}), /server rejected/); + assert.equal(fixture.calls.length, 1); + assert.deepEqual(fixture.calls[0]!.params._meta?.[CLIENT_CAPABILITIES_META_KEY], {}); +}); + +test('multiple rounds replace old responses and preserve exact opaque state', async () => { + const { fixture, manager, binding } = await setup(); + const first = ' round-one \n opaque\u0000 '; + const second = 'round-two-state'; + fixture.respond = () => + fixture.calls.length === 1 + ? inputRequired({ + inputRequests: { z: mcpFixtureFormRequest(), a: mcpFixtureFormRequest() }, + requestState: first, + }) + : fixture.calls.length === 2 + ? inputRequired({ inputRequests: { next: mcpFixtureFormRequest() }, requestState: second }) + : { content: [] }; + let forms = 0; + await manager.callTool( + binding, + {}, + { + requestInteraction: async () => { + forms++; + return accepted; + }, + }, + ); + assert.equal(forms, 3); + assert.equal(fixture.calls.length, 3); + assert.equal(fixture.calls[1]!.params.requestState, first); + assert.deepEqual(Object.keys(fixture.calls[1]!.params.inputResponses!), ['a', 'z']); + assert.equal(fixture.calls[2]!.params.requestState, second); + assert.deepEqual(Object.keys(fixture.calls[2]!.params.inputResponses!), ['next']); + assert.equal(new Set(fixture.calls.map(({ id }) => id)).size, 3); +}); + +test('state-only rounds are bounded and retain an empty state exactly', async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = () => inputRequired({ requestState: '' }); + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => assert.fail('unexpected form'), + }, + ), + /round limit/, + ); + assert.equal(fixture.calls.length, 9); + assert.equal(fixture.calls[1]!.params.requestState, ''); + assert.equal(Object.hasOwn(fixture.calls[1]!.params, 'inputResponses'), false); +}); + +test('invalid sibling fails preflight without displaying the valid first form', async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = () => + inputRequired({ + inputRequests: { + first: mcpFixtureFormRequest(), + second: { + ...mcpFixtureFormRequest(), + params: { ...mcpFixtureFormRequest().params, message: 'm'.repeat(2_049) }, + }, + }, + requestState: 'opaque', + }); + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => assert.fail('preflight was not atomic'), + }, + ), + ); + assert.equal(fixture.calls.length, 1); +}); + +test('invalid accepted answer never reaches server', async () => { + const { fixture, manager, binding } = await setup(); + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => ({ + action: 'accept', + values: { name: 'Ada', email: 'invalid', confirm: true }, + }), + }, + ), + ); + assert.equal(fixture.calls.length, 1); +}); + +for (const invalidate of [ + 'stop', + 'disconnect', + 'reconnect', + 'close', + 'changed refresh', + 'exhaust', +] as const) { + test(`${invalidate} aborts a pending callback even when it ignores cancellation`, async () => { + const { fixture, manager, binding } = await setup(); + const shown = deferred(); + const answer = deferred(); + const controller = new AbortController(); + const call = manager.callTool( + binding, + {}, + { + signal: controller.signal, + requestInteraction: async () => { + shown.resolve(); + return answer.promise; + }, + }, + ); + const rejected = assert.rejects( + call, + invalidate === 'exhaust' ? /retention exhausted/ : /aborted|stale/, + ); + await shown.promise; + if (invalidate === 'stop') controller.abort(); + else if (invalidate === 'disconnect') await manager.disconnect('forms'); + else if (invalidate === 'reconnect') await manager.reconnect('forms'); + else if (invalidate === 'close') await manager.close(); + else if (invalidate === 'exhaust') { + fixture.respond = () => inputRequired({ requestState: 's'.repeat(16 * 1024 + 1) }); + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => assert.fail('exhausting call must not publish a form'), + }, + ), + /retention exhausted/, + ); + } else { + fixture.definition = { ...fixture.definition, description: 'new definition' }; + await manager.refreshTools('forms'); + } + await Promise.race([ + rejected, + delay(500).then(() => assert.fail('pending invocation did not abort')), + ]); + answer.resolve(accepted); + await delay(10); + assert.equal(fixture.calls.length, invalidate === 'exhaust' ? 2 : 1); + }); +} + +test('unchanged refresh preserves the pending invocation', async () => { + const { fixture, manager, binding } = await setup(); + await manager.callTool( + binding, + {}, + { + requestInteraction: async () => { + await manager.refreshTools('forms'); + assert.equal(manager.toolSnapshot().tools[0]!.binding, binding); + return accepted; + }, + }, + ); + assert.equal(fixture.calls.length, 2); +}); + +test('abort just after answer commitment prevents the next network leg', async () => { + const { fixture, manager, binding } = await setup(); + const controller = new AbortController(); + await assert.rejects( + manager.callTool( + binding, + {}, + { + signal: controller.signal, + requestInteraction: async () => { + controller.abort(); + return accepted; + }, + }, + ), + /aborted/, + ); + assert.equal(fixture.calls.length, 1); +}); + +test('user wait does not consume the network timeout', async () => { + const { fixture, manager, binding } = await setup(); + await manager.callTool( + binding, + {}, + { + timeoutMs: 100, + requestInteraction: async () => { + await delay(150); + return accepted; + }, + }, + ); + assert.equal(fixture.calls.length, 2); +}); + +test('separate invocations on one connection keep answers and states isolated', async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = (params) => + params.inputResponses + ? { content: [{ type: 'text', text: String(params.inputResponses.form?.action) }] } + : inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: `state-${params.arguments?.id}`, + }); + await Promise.all( + ['left', 'right'].map((id) => + manager.callTool( + binding, + { id }, + { + requestInteraction: async () => ({ action: id === 'left' ? 'decline' : 'cancel' }), + }, + ), + ), + ); + const retries = fixture.calls.filter(({ params }) => params.inputResponses); + assert.equal(retries.length, 2); + for (const { params } of retries) { + assert.equal(params.requestState, `state-${params.arguments?.id}`); + assert.equal( + params.inputResponses!.form?.action, + params.arguments?.id === 'left' ? 'decline' : 'cancel', + ); + } +}); + +for (const state of ['private-long-state', '§']) { + for (const outcome of ['success', 'tool-error', 'protocol-error'] as const) { + test(`scrubs ${state.length > 1 ? 'long' : 'short'} state reflected in ${outcome}`, async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = (params) => { + if (!params.inputResponses) + return inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: state, + }); + if (outcome === 'protocol-error') + throw new ProtocolError(-32000, `reflected ${state}`, { state }); + return { + content: [{ type: 'text', text: `reflected ${state}` }], + structuredContent: { echoed: state }, + ...(outcome === 'tool-error' ? { isError: true } : {}), + }; + }; + if (outcome === 'success') { + const result = await manager.callTool( + binding, + {}, + { requestInteraction: async () => accepted }, + ); + assert.equal(JSON.stringify(result).includes(state), false); + } else { + await assert.rejects( + manager.callTool(binding, {}, { requestInteraction: async () => accepted }), + (error: unknown) => { + assert(error instanceof Error); + assert.equal(error.message.includes(state), false); + if (error.cause instanceof Error) + assert.equal(error.cause.message.includes(state), false); + assert.equal(JSON.stringify(error).includes(state), false); + return true; + }, + ); + } + }); + } +} + +for (const state of ['opaque-secret-state', 'opaque\nstate', '[redacted]']) { + for (const location of ['message', 'property key', 'default'] as const) { + test(`state ${JSON.stringify(state)} echoed into a form ${location} fails before projection`, async () => { + const { fixture, manager, binding } = await setup(); + const request = mcpFixtureFormRequest(); + if (location === 'message') request.params.message = state; + else if (location === 'property key') { + Object.defineProperty(request.params.requestedSchema.properties, state, { + value: { type: 'string' }, + enumerable: true, + }); + } else { + Object.assign(request.params.requestedSchema.properties.name, { default: state }); + } + fixture.respond = (params) => + params.inputResponses + ? { content: [] } + : inputRequired({ inputRequests: { form: request }, requestState: state }); + let shown = 0; + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => { + shown++; + return accepted; + }, + }, + ), + /private continuation/, + ); + assert.equal(shown, 0); + assert.equal(fixture.calls.length, 1); + }); + } +} + +for (const state of ['opaque\nstate', '[redacted]']) { + test(`a later call cannot project earlier connection state ${JSON.stringify(state)}`, async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = (params) => + params.inputResponses + ? { content: [] } + : inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: state, + }); + await manager.callTool(binding, {}, { requestInteraction: async () => accepted }); + const request = mcpFixtureFormRequest(); + request.params.message = state; + fixture.respond = (params) => + params.inputResponses + ? { content: [] } + : inputRequired({ + inputRequests: { form: request }, + requestState: 'new-private-state', + }); + let shown = 0; + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => { + shown++; + return accepted; + }, + }, + ), + /private continuation/, + ); + assert.equal(shown, 0); + assert.equal(fixture.calls.length, 3); + }); +} + +test('retained state cannot escape through a later output-schema preparation error', async () => { + const { fixture, manager, binding } = await setup(); + await manager.callTool(binding, {}, { requestInteraction: async () => accepted }); + fixture.definition = { + ...fixture.definition, + outputSchema: { type: 'object', $ref: 'opaque-state' }, + }; + await manager.refreshTools('forms'); + const refreshed = manager.toolSnapshot().tools[0]!.binding; + await assert.rejects(manager.callTool(refreshed, {}), (error: unknown) => { + assert(error instanceof Error); + assert.match(error.message, /invalid output schema/); + assert(error.cause instanceof Error); + assert.doesNotMatch(error.cause.message, /opaque-state/); + return true; + }); + assert.equal(fixture.calls.length, 2); +}); + +test('retained state cannot escape through a later header-argument error', async () => { + const { fixture, manager, binding } = await setup(); + await manager.callTool(binding, {}, { requestInteraction: async () => accepted }); + fixture.definition = { + ...fixture.definition, + inputSchema: { + type: 'object', + properties: { 'opaque-state': { type: 'integer', 'x-mcp-header': 'Shard' } }, + }, + }; + await manager.refreshTools('forms'); + const refreshed = manager.toolSnapshot().tools[0]!.binding; + await assert.rejects( + manager.callTool(refreshed, { 'opaque-state': Number.MAX_SAFE_INTEGER + 1 }), + (error: unknown) => { + assert(error instanceof Error); + assert.match(error.message, /unsafe integer/); + assert.doesNotMatch(error.message, /opaque-state/); + return true; + }, + ); + assert.equal(fixture.calls.length, 2); +}); + +for (const interactive of [false, true]) { + test(`retention exhaustion fences an already pending ${interactive ? 'interactive' : 'ordinary'} call`, async () => { + const { fixture, manager, binding } = await setup(); + const waiting = deferred(); + const release = deferred(); + let stateNumber = 0; + fixture.respond = async (params) => { + if (params.arguments?.hold) { + waiting.resolve(); + await release.promise; + return { content: [{ type: 'text', text: 'state-65' }] }; + } + return params.inputResponses + ? { content: [] } + : inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: `state-${++stateNumber}`, + }); + }; + for (let call = 0; call < 64; call++) { + await manager.callTool(binding, {}, { requestInteraction: async () => accepted }); + } + const pending = assert.rejects( + manager.callTool( + binding, + { hold: true }, + { + ...(interactive ? { requestInteraction: async () => accepted } : {}), + }, + ), + /retention exhausted/, + ); + await waiting.promise; + try { + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => accepted, + }, + ), + /retention exhausted/, + ); + // Cancellation must settle the call while the server is still blocked; + // releasing first would also pass if only the post-response fence worked. + await Promise.race([ + pending, + delay(500).then(() => assert.fail('pending request was not promptly cancelled')), + ]); + } finally { + release.resolve(); + } + const requests = fixture.calls.length; + await assert.rejects(manager.callTool(binding, {}), /retention exhausted/); + assert.equal(fixture.calls.length, requests); + }); +} + +for (const limit of ['state', 'count', 'bytes'] as const) { + test(`rejects excessive ${limit} before displaying a form`, async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = () => + inputRequired({ + requestState: limit === 'state' ? 's'.repeat(16 * 1024 + 1) : 'opaque', + inputRequests: + limit === 'count' + ? Object.fromEntries( + Array.from({ length: 9 }, (_, index) => [String(index), mcpFixtureFormRequest()]), + ) + : { + form: { + ...mcpFixtureFormRequest(), + params: { + ...mcpFixtureFormRequest().params, + message: limit === 'bytes' ? 'm'.repeat(64 * 1024) : 'Please fill', + }, + }, + }, + }); + await assert.rejects( + manager.callTool( + binding, + {}, + { + requestInteraction: async () => assert.fail('over-limit form reached callback'), + }, + ), + /limit/, + ); + assert.equal(fixture.calls.length, 1); + }); +} + +test('the retry network leg still has its own timeout', async () => { + const { fixture, manager, binding } = await setup(); + const respond = fixture.respond; + fixture.respond = async (params) => { + if (params.inputResponses) await delay(250); + return respond(params); + }; + await assert.rejects( + manager.callTool( + binding, + {}, + { + timeoutMs: 100, + requestInteraction: async () => accepted, + }, + ), + /timed out/, + ); + assert.equal(fixture.calls.length, 2); +}); + +test('old state is scrubbed even after a later round replaces it', async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = () => + fixture.calls.length === 1 + ? inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: 'first-private-state', + }) + : fixture.calls.length === 2 + ? inputRequired({ + inputRequests: { next: mcpFixtureFormRequest() }, + requestState: 'second-private-state', + }) + : { content: [{ type: 'text', text: 'first-private-state second-private-state' }] }; + const result = await manager.callTool(binding, {}, { requestInteraction: async () => accepted }); + assert.doesNotMatch(JSON.stringify(result), /first-private-state|second-private-state/); +}); + +test('a new round without state does not reuse previous state', async () => { + const { fixture, manager, binding } = await setup(); + fixture.respond = () => + fixture.calls.length === 1 + ? inputRequired({ + inputRequests: { form: mcpFixtureFormRequest() }, + requestState: 'first-private-state', + }) + : fixture.calls.length === 2 + ? inputRequired({ inputRequests: { next: mcpFixtureFormRequest() } }) + : { content: [] }; + await manager.callTool(binding, {}, { requestInteraction: async () => accepted }); + assert.equal(Object.hasOwn(fixture.calls[2]!.params, 'requestState'), false); +}); + +test('interactive retry retains SEP-2243 headers and final output validation', async () => { + const { fixture, manager } = await setup(); + fixture.definition = { + ...fixture.definition, + inputSchema: { + type: 'object', + properties: { shard: { type: 'integer', 'x-mcp-header': 'Shard' } }, + }, + outputSchema: { type: 'object', properties: { done: { type: 'boolean' } }, required: ['done'] }, + }; + await manager.refreshTools('forms'); + const binding = manager.toolSnapshot().tools[0]!.binding; + // The initial deferred result has no structuredContent. Only the complete + // result is subject to the original frozen tool's output schema. + await assert.rejects( + manager.callTool( + binding, + { shard: 42 }, + { + requestInteraction: async () => accepted, + }, + ), + /invalid tool result/, + ); + assert.equal(fixture.calls.length, 2); + assert.deepEqual( + fixture.callHeaders.map((headers) => headers['mcp-param-shard']), + ['42', '42'], + ); +}); + +test('a supplied form callback never advertises elicitation on a legacy connection', async () => { + const { fixture, manager, binding } = await setup(true); + fixture.respond = () => ({ content: [{ type: 'text', text: 'legacy complete' }] }); + assert.equal(manager.status('forms')?.negotiatedProtocol?.era, 'legacy'); + await manager.callTool( + binding, + {}, + { + requestInteraction: async () => assert.fail('legacy callback must remain disabled'), + }, + ); + assert.deepEqual(fixture.initializeCapabilities, {}); + assert.equal(fixture.calls[0]!.params._meta?.[CLIENT_CAPABILITIES_META_KEY], undefined); + assert.equal(fixture.calls.length, 1); +}); diff --git a/packages/mcp/src/__tests__/form-stdio.test.ts b/packages/mcp/src/__tests__/form-stdio.test.ts new file mode 100644 index 0000000000..1c2ec72f05 --- /dev/null +++ b/packages/mcp/src/__tests__/form-stdio.test.ts @@ -0,0 +1,215 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { fileURLToPath } from 'node:url'; +import { test } from 'node:test'; +import { MCP_CONFIG_VERSION } from '@maka/core/mcp'; +import { McpClientManager } from '../index.js'; + +const fixturePath = fileURLToPath(new URL('../__fixtures__/form-stdio-server.js', import.meta.url)); + +test('modern-only stdio auto negotiation completes a form and protects state reflected on stderr', async () => { + const manager = new McpClientManager({ timeouts: { stdioConnectMs: 5_000, callToolMs: 2_000 } }); + const statuses: unknown[] = []; + manager.onChange((status) => statuses.push(status)); + try { + await manager.sync({ + version: MCP_CONFIG_VERSION, + mcpServers: { + stdio: { command: process.execPath, args: [fixturePath], protocol: 'auto' }, + }, + }); + assert.deepEqual(manager.status('stdio')?.negotiatedProtocol, { + era: 'modern', + revision: '2026-07-28', + }); + const binding = manager.toolSnapshot().tools[0]?.binding; + assert.ok(binding); + let forms = 0; + const values = { name: 'Ada', email: 'ada@example.com', confirm: true }; + const result = await manager.callTool( + binding, + {}, + { + requestInteraction: async (form) => { + forms += 1; + assert.deepEqual( + form.fields.map((field) => field.name), + ['name', 'email', 'confirm'], + ); + return { action: 'accept', values }; + }, + }, + ); + assert.equal(forms, 1); + assert.deepEqual(result.structuredContent, { answer: { action: 'accept', content: values } }); + await waitForDiagnostic(manager, 'safe diagnostic'); + assert.ok(manager.status('stdio')?.stderrTail?.some((line) => line.includes('[redacted]'))); + assert.equal( + JSON.stringify([statuses, manager.status('stdio'), result]).includes( + 'stdio-private-continuation-state', + ), + false, + ); + } finally { + await manager.close(); + } +}); + +const PRIVATE_STATE = 'stdio-private-continuation-state'; +const FORM_ANSWER = { + action: 'accept' as const, + values: { name: 'Ada', email: 'ada@example.com', confirm: true }, +}; + +for (const mode of ['before', 'partial', 'continued', 'control'] as const) { + test(`stdio ${mode} cannot expose continuation state before response or after settlement`, { + timeout: 10_000, + }, async () => { + await withStdio(async (manager, statuses) => { + const binding = manager.toolSnapshot().tools[0]!.binding; + await manager.callTool(binding, { mode }, { requestInteraction: async () => FORM_ANSWER }); + if (mode === 'control') { + await new Promise((resolve) => setTimeout(resolve, 100)); + await manager.callTool(binding, { mode: 'diagnostic' }); + assert.deepEqual(manager.status('stdio')?.stderrTail ?? [], []); + } else { + await waitForDiagnostic(manager, 'safe diagnostic'); + } + const published = JSON.stringify([statuses, manager.status('stdio')]); + assert.equal(published.includes(PRIVATE_STATE), false); + assert.equal(published.includes('control-private-suffix'), false); + assert.equal(published.includes(PRIVATE_STATE.slice(12)), false); + }); + }); +} + +for (const [mode, allowed] of [ + ['retention', 64], + ['byte-retention', 16], +] as const) { + test(`stdio ${mode} retains old states without eviction and fails closed until reconnect`, { + timeout: 20_000, + }, async () => { + await withStdio(async (manager, statuses) => { + const binding = manager.toolSnapshot().tools[0]!.binding; + for (let call = 0; call < allowed; call += 1) { + await manager.callTool(binding, { mode }, { requestInteraction: async () => FORM_ANSWER }); + } + await manager.callTool(binding, { mode: 'first-diagnostic' }); + assert.equal(JSON.stringify(statuses).includes(`${PRIVATE_STATE}-1`), false); + await assert.rejects( + manager.callTool(binding, { mode }, { requestInteraction: async () => FORM_ANSWER }), + ); + await assert.rejects( + manager.callTool( + binding, + { mode }, + { + requestInteraction: async () => + assert.fail('exhausted connection must not show a form'), + }, + ), + /retention exhausted/, + ); + await assert.rejects( + manager.callTool(binding, { mode: 'diagnostic' }), + /retention exhausted/, + ); + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.deepEqual(manager.status('stdio')?.stderrTail ?? [], []); + assert.equal(JSON.stringify(statuses).includes(PRIVATE_STATE), false); + await manager.reconnect('stdio'); + const freshBinding = manager.toolSnapshot().tools[0]!.binding; + await manager.callTool(freshBinding, {}, { requestInteraction: async () => FORM_ANSWER }); + await waitForDiagnostic(manager, 'safe diagnostic'); + assert.equal( + JSON.stringify([statuses, manager.status('stdio')]).includes(PRIVATE_STATE), + false, + ); + }); + }); +} + +async function withStdio(run: (manager: McpClientManager, statuses: unknown[]) => Promise) { + const manager = new McpClientManager({ timeouts: { stdioConnectMs: 5_000, callToolMs: 2_000 } }); + const statuses: unknown[] = []; + manager.onChange((status) => statuses.push(status)); + try { + await manager.sync({ + version: MCP_CONFIG_VERSION, + mcpServers: { + stdio: { command: process.execPath, args: [fixturePath], protocol: 'auto' }, + }, + }); + await run(manager, statuses); + } finally { + await manager.close(); + } +} + +async function waitForDiagnostic(manager: McpClientManager, text: string): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (manager.status('stdio')?.stderrTail?.some((line) => line.includes(text))) return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.fail(`Expected stderr diagnostic: ${text}`); +} + +test('stdio non-ASCII state suppresses the generation even when UTF-8 stderr bytes split', { + timeout: 10_000, +}, async () => { + await withStdio(async (manager, statuses) => { + const binding = manager.toolSnapshot().tools[0]!.binding; + await manager.callTool( + binding, + { mode: 'unicode' }, + { requestInteraction: async () => FORM_ANSWER }, + ); + await manager.callTool(binding, { mode: 'unicode-diagnostic' }); + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.deepEqual(manager.status('stdio')?.stderrTail ?? [], []); + assert.equal(JSON.stringify(statuses).includes(PRIVATE_STATE), false); + }); +}); + +test('learning state clears already formatted or truncated stderr and notifies subscribers', { + timeout: 10_000, +}, async () => { + await withStdio(async (manager, statuses) => { + const binding = manager.toolSnapshot().tools[0]!.binding; + await manager.callTool(binding, { mode: 'future-diagnostic' }); + await waitForDiagnostic(manager, 'future-state-prefix'); + // The server emitted this value before it was identified as private state. + assert.equal(JSON.stringify(statuses).includes('future-state-prefix'), true); + await manager.callTool( + binding, + { mode: 'future' }, + { + requestInteraction: async () => { + assert.deepEqual(manager.status('stdio')?.stderrTail ?? [], []); + assert.equal(JSON.stringify(statuses.at(-1)).includes('future-state-prefix'), false); + return FORM_ANSWER; + }, + }, + ); + assert.equal(JSON.stringify(manager.status('stdio')).includes('future-state-prefix'), false); + }); +}); diff --git a/packages/mcp/src/form-elicitation.ts b/packages/mcp/src/form-elicitation.ts new file mode 100644 index 0000000000..3140142f46 --- /dev/null +++ b/packages/mcp/src/form-elicitation.ts @@ -0,0 +1,233 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { + decodeInteractionAnswer, + decodeInteractionRequest, + interactionFormAnswerMatchesRequest, + type InteractionFormInput, + type InteractionFormResult, + projectInteractionFormRequest, +} from '@maka/core/interaction'; +import type { ElicitResult } from '@modelcontextprotocol/client'; + +// Unbounded protocol strings need a visible client input limit: Core reserves +// the worst-case JSON-escaped answer within its serialized interaction budget. +// Explicit server bounds are preserved and still checked by Core admission. +const DEFAULT_STRING_MAX_LENGTH = 256; + +/** Inspect raw keys and values before projection can escape control characters. + * This is a containment check, not a comparison with redacted output: a state + * may itself be identical to the redaction marker. + */ +export function containsMcpFormState(value: unknown, states: readonly string[]): boolean { + const pending: unknown[] = [value]; + while (pending.length) { + const item = pending.pop(); + if (typeof item === 'string') { + if (states.some((state) => state.length > 0 && item.includes(state))) return true; + } else if (item !== null && typeof item === 'object') { + for (const [key, child] of Object.entries(item)) pending.push(key, child); + } + } + return false; +} + +function record(value: unknown): Record { + if (value === null || typeof value !== 'object' || Array.isArray(value)) throw new Error(); + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) throw new Error(); + return value as Record; +} + +function keys(value: Record, allowed: readonly string[]): void { + if (Object.keys(value).some((key) => !allowed.includes(key))) throw new Error(); +} + +function select(value: Record, names: readonly string[]): Record { + return Object.fromEntries( + names.filter((name) => Object.hasOwn(value, name)).map((name) => [name, value[name]]), + ); +} + +function options(value: unknown, titles?: unknown): unknown[] { + if ( + !Array.isArray(value) || + (titles !== undefined && (!Array.isArray(titles) || titles.length !== value.length)) + ) + throw new Error(); + return value.map((entry, index) => ({ + value: entry, + label: Array.isArray(titles) ? titles[index] : entry, + })); +} + +function titledOptions(value: unknown): unknown[] { + if (!Array.isArray(value)) throw new Error(); + return value.map((entry) => { + const option = record(entry); + keys(option, ['const', 'title']); + return { value: option.const, label: option.title }; + }); +} + +function field(name: string, value: unknown, required: boolean): Record { + const schema = record(value); + const sharedKeys = ['type', 'title', 'description', 'default']; + const shared = { + name, + label: Object.hasOwn(schema, 'title') ? schema.title : name, + required, + ...select(schema, ['description', 'default']), + }; + if (schema.type === 'string') { + if (Object.hasOwn(schema, 'enum')) { + keys(schema, [...sharedKeys, 'enum', 'enumNames']); + return { ...shared, kind: 'single_select', options: options(schema.enum, schema.enumNames) }; + } + if (Object.hasOwn(schema, 'oneOf')) { + keys(schema, [...sharedKeys, 'oneOf']); + return { ...shared, kind: 'single_select', options: titledOptions(schema.oneOf) }; + } + keys(schema, [...sharedKeys, 'minLength', 'maxLength', 'format']); + return { + ...shared, + kind: 'string', + ...select(schema, ['minLength', 'maxLength', 'format']), + ...(schema.maxLength === undefined ? { maxLength: DEFAULT_STRING_MAX_LENGTH } : {}), + }; + } + if (schema.type === 'number' || schema.type === 'integer') { + keys(schema, [...sharedKeys, 'minimum', 'maximum']); + // Core validates numeric ranges; an integer-only interval may additionally be empty. + if ( + schema.type === 'integer' && + typeof schema.minimum === 'number' && + typeof schema.maximum === 'number' && + Math.ceil(schema.minimum) > Math.floor(schema.maximum) + ) + throw new Error(); + return { ...shared, kind: schema.type, ...select(schema, ['minimum', 'maximum']) }; + } + if (schema.type === 'boolean') { + keys(schema, sharedKeys); + return { ...shared, kind: 'boolean' }; + } + if (schema.type === 'array') { + keys(schema, [...sharedKeys, 'items', 'minItems', 'maxItems']); + const items = record(schema.items); + let choices: unknown[]; + if (Object.hasOwn(items, 'anyOf')) { + keys(items, ['anyOf']); + choices = titledOptions(items.anyOf); + } else { + keys(items, ['type', 'enum']); + if (items.type !== 'string') throw new Error(); + choices = options(items.enum); + } + return { + ...shared, + kind: 'multi_select', + options: choices, + ...select(schema, ['minItems', 'maxItems']), + }; + } + throw new Error(); +} + +/** Adapt one embedded MCP form request without exposing protocol state to the UI. */ +export function prepareMcpForm( + request: unknown, + requester: { name: string; source?: string }, +): { form: InteractionFormInput; respond(result: InteractionFormResult): ElicitResult } { + try { + const embedded = record(request); + keys(embedded, ['method', 'params']); + if (embedded.method !== 'elicitation/create') throw new Error(); + const params = record(embedded.params); + keys(params, ['mode', 'message', 'requestedSchema', '_meta']); + if (params.mode !== undefined && params.mode !== 'form') throw new Error(); + const schema = record(params.requestedSchema); + keys(schema, [ + 'type', + 'properties', + 'required', + '$schema', + 'title', + 'description', + 'additionalProperties', + ]); + if ( + schema.type !== 'object' || + (schema.additionalProperties !== undefined && schema.additionalProperties !== false) + ) + throw new Error(); + for (const annotation of ['$schema', 'title', 'description']) { + if (schema[annotation] !== undefined && typeof schema[annotation] !== 'string') + throw new Error(); + } + const properties = record(schema.properties); + const required: unknown = schema.required === undefined ? [] : schema.required; + if ( + !Array.isArray(required) || + required.some((name) => typeof name !== 'string' || !Object.hasOwn(properties, name)) || + new Set(required).size !== required.length + ) + throw new Error(); + const decoded = decodeInteractionRequest({ + kind: 'form', + toolUseId: 'mcp-form', + message: params.message, + requester, + fields: Object.entries(properties).map(([name, schema]) => + field(name, schema, required.includes(name)), + ), + }); + if (decoded.kind !== 'form') throw new Error(); + const projected = projectInteractionFormRequest(decoded); + return { + form: { + message: projected.message, + requester: projected.requester, + fields: projected.fields, + }, + respond(result) { + try { + const answer = decodeInteractionAnswer({ ...result, kind: 'form' }); + if (answer.kind !== 'form' || !interactionFormAnswerMatchesRequest(projected, answer)) + throw new Error(); + if (answer.action !== 'accept') return { action: answer.action }; + return { + action: 'accept', + content: Object.fromEntries( + Object.entries(answer.values).map(([name, value]) => [ + name, + typeof value === 'object' ? [...value] : value, + ]), + ), + }; + } catch { + throw new Error('Invalid MCP form response'); + } + }, + }; + } catch { + throw new Error('Unsupported or invalid MCP form request'); + } +} diff --git a/packages/mcp/src/index.ts b/packages/mcp/src/index.ts index ac435180bb..4d7b35d773 100644 --- a/packages/mcp/src/index.ts +++ b/packages/mcp/src/index.ts @@ -23,6 +23,9 @@ import { Client, extractWWWAuthenticateParams, LATEST_PROTOCOL_VERSION, + CLIENT_CAPABILITIES_META_KEY, + isInputRequiredResult, + type ElicitResult, SdkErrorCode, SdkHttpError, SSEClientTransport, @@ -34,6 +37,9 @@ import { type VersionNegotiationOptions, } from '@modelcontextprotocol/client'; import { StdioClientTransport } from '@modelcontextprotocol/client/stdio'; +import type { InteractionFormInput, InteractionFormResult } from '@maka/core/interaction'; +import { setTimeout as delay } from 'node:timers/promises'; +import { containsMcpFormState, prepareMcpForm } from './form-elicitation.js'; import { redactSecrets } from '@maka/core/redaction'; import { serializedByteLength } from '@maka/core/serialized-byte-length'; import { @@ -145,6 +151,20 @@ const TOOL_REFRESH_BURST_IDLE_MS = 1_000; // access/refresh/id token, client secret, verifier), small enough that // deepScrub stays O(bound) per payload for the life of the process. const MAX_HARVESTED_SECRETS_PER_SERVER = 40; +// No eviction: a live transport may reflect any previous continuation later. +const MAX_CONNECTION_FORM_STATES = 64; +const MAX_CONNECTION_FORM_STATE_BYTES = 256 * 1024; +const FORM_PRIVACY_EXHAUSTED = 'form privacy retention exhausted (state limit); reconnect required'; + +interface ConnectionDiagnosticPrivacy { + readonly states: Set; + readonly failure: AbortController; + bytes: number; + pendingCalls: number; + stderrSuppressed: boolean; + exhausted: boolean; + discardPendingStderr?: () => void; +} export interface McpClientManagerOptions { clientName?: string; @@ -217,6 +237,7 @@ interface ToolRefreshNotificationState { } interface Connection { + diagnosticPrivacy?: ConnectionDiagnosticPrivacy; config: McpServerConfig; fingerprint: string; /** Set when stored credentials for this entry's (old) config could not be @@ -267,6 +288,14 @@ export class McpClientManager { * pass — and racing one against the round trips the round's version * fence, failing the user's login over nothing they did. */ private readonly interactiveRounds = new Set(); + private readonly formCalls = new Map< + AbortController, + { + binding: McpToolBinding; + serverId: string; + states: string[]; + } + >(); private bindingIndex = new Map(); private readonly listeners = new Set(); private syncQueue: Promise = Promise.resolve(); @@ -387,6 +416,25 @@ export class McpClientManager { value, ); } + const privacy = this.connections.get(serverId)?.diagnosticPrivacy; + if (privacy?.exhausted) { + // An empty withheld token intentionally matches every diagnostic. Once + // retention is full, fail closed until a fresh physical connection. + inventory.withhold.push(''); + } + for (const state of privacy?.states ?? []) { + (state.length >= MIN_SUBSTITUTION_LENGTH ? inventory.substitute : inventory.withhold).push( + state, + ); + } + for (const call of this.formCalls.values()) { + if (call.serverId !== serverId) continue; + for (const state of call.states) { + (state.length >= MIN_SUBSTITUTION_LENGTH ? inventory.substitute : inventory.withhold).push( + state, + ); + } + } return inventory; } @@ -753,7 +801,14 @@ export class McpClientManager { async callTool( binding: McpToolBinding, args: Record, - options: { signal?: AbortSignal; timeoutMs?: number } = {}, + options: { + signal?: AbortSignal; + timeoutMs?: number; + requestInteraction?: ( + form: InteractionFormInput, + options?: { cancellationSignal?: AbortSignal }, + ) => Promise; + } = {}, ): Promise { const identity = parseMcpToolBinding(binding); if (!identity) { @@ -783,111 +838,269 @@ export class McpClientManager { throw new McpToolCallError(serverId, toolName, 'tool binding is stale'); } const client = entry.client; + let inventory = this.secretsFor(serverId, entry.config); const headerArguments = validateMcpHeaderArguments(snapshot.headerDeclarations, args); if (!headerArguments.valid) { throw new McpToolCallError( serverId, toolName, - `unsafe integer argument at ${formatMcpDiagnosticText(headerArguments.path.join('.'))}`, + `unsafe integer argument at ${formatMcpDiagnosticText(scrubKnownSecrets(headerArguments.path.join('.'), inventory))}`, ); } - const inventory = this.secretsFor(serverId, entry.config); const preparation = snapshot.callPreparation ?? (snapshot.callPreparation = this.toolCallPreparer.prepare(snapshot.definition)); if (!preparation.ok) { throw new McpToolCallError(serverId, toolName, 'server advertised an invalid output schema', { - cause: preparation.cause, + cause: sanitizedCause(preparation.cause, inventory), }); } - let result; + const requestInteraction = + entry.status.negotiatedProtocol?.era === 'modern' ? options.requestInteraction : undefined; + const privacy = entry.diagnosticPrivacy; + if (privacy?.exhausted) { + throw new McpToolCallError(serverId, toolName, FORM_PRIVACY_EXHAUSTED); + } + const controller = requestInteraction ? new AbortController() : undefined; + const signal = controller + ? AbortSignal.any([ + controller.signal, + ...(privacy ? [privacy.failure.signal] : []), + ...(options.signal ? [options.signal] : []), + ]) + : privacy + ? AbortSignal.any([privacy.failure.signal, ...(options.signal ? [options.signal] : [])]) + : options.signal; + const refreshInventory = () => { + inventory = this.secretsFor(serverId, entry.config); + if (privacy?.exhausted) inventory.withhold.push(''); + for (const state of privacy?.states ?? []) { + (state.length >= MIN_SUBSTITUTION_LENGTH ? inventory.substitute : inventory.withhold).push( + state, + ); + } + }; + const states: string[] = []; + if (controller) { + this.formCalls.set(controller, { binding, serverId, states }); + if (privacy) { + privacy.pendingCalls += 1; + privacy.discardPendingStderr?.(); + } + } + const assertCurrent = () => { + signal?.throwIfAborted(); + if ( + this.closed || + entry.closing || + this.connections.get(serverId) !== entry || + entry.client !== client || + entry.status.state !== 'connected' || + entry.connectionGeneration !== identity.connectionGeneration || + entry.toolSnapshot.get(toolName)?.binding !== binding + ) + throw new McpToolCallError(serverId, toolName, 'tool binding is stale'); + }; try { - result = await client.callTool( - { name: toolName, arguments: args }, - { - signal: options.signal, - timeout: options.timeoutMs ?? this.timeouts.callToolMs, - toolDefinition: structuredClone(preparation.value.definitionForSdk), - }, - ); + const originalArguments = requestInteraction ? structuredClone(args) : args; + let continuation: { inputResponses?: Record; requestState?: string } = + {}; + let rounds = 0; + let result; + while (true) { + if (requestInteraction) assertCurrent(); + result = await client.callTool( + { + name: toolName, + arguments: originalArguments, + ...continuation, + ...(requestInteraction + ? { + _meta: { [CLIENT_CAPABILITIES_META_KEY]: { elicitation: { form: {} } } }, + } + : {}), + }, + { + signal, + timeout: options.timeoutMs ?? this.timeouts.callToolMs, + toolDefinition: structuredClone(preparation.value.definitionForSdk), + ...(requestInteraction ? { allowInputRequired: true } : {}), + }, + ); + refreshInventory(); + if (privacy?.exhausted) + throw new McpToolCallError(serverId, toolName, FORM_PRIVACY_EXHAUSTED); + if (requestInteraction) assertCurrent(); + if (!requestInteraction || !isInputRequiredResult(result)) break; + const state = result.requestState; + if ( + state !== undefined && + (typeof state !== 'string' || Buffer.byteLength(state) > 16 * 1024) + ) { + if (privacy) { + exhaustConnectionDiagnosticPrivacy(entry, privacy); + this.emit(entry.status); + } + inventory.withhold.push(''); + throw new McpToolCallError( + serverId, + toolName, + 'form continuation state exceeds the limit', + ); + } + const learnedState = Boolean(state && privacy && !privacy.states.has(state)); + if (state) { + if (privacy && !retainConnectionFormState(entry, privacy, state)) { + this.emit(entry.status); + inventory.withhold.push(''); + throw new McpToolCallError(serverId, toolName, FORM_PRIVACY_EXHAUSTED); + } + states.push(state); + (state.length >= MIN_SUBSTITUTION_LENGTH + ? inventory.substitute + : inventory.withhold + ).push(state); + } + if (learnedState && entry.status.stderrTail !== undefined) { + // Existing lines have already been formatted/truncated, so an exact + // replacement cannot reliably remove fragments of newly learned + // state. Clear the cache and notify consumers. Earlier published + // diagnostics from before the state was known cannot be retracted. + entry.status = { ...entry.status, stderrTail: undefined }; + this.emit(entry.status); + } + if (++rounds > 8) + throw new McpToolCallError(serverId, toolName, 'form round limit exceeded'); + const requests = result.inputRequests ?? {}; + const keys = Object.keys(requests).sort(compareText); + if (keys.length > 8 || serializedByteLength(requests, 64 * 1024) > 64 * 1024) { + throw new McpToolCallError(serverId, toolName, 'form input requests exceed the limit'); + } + if (containsMcpFormState(requests, privacy ? [...privacy.states] : states)) { + throw new McpToolCallError( + serverId, + toolName, + 'form contains private continuation material', + ); + } + // Validate every sibling before publishing any user-facing Interaction. + const forms = keys.map((key) => ({ + key, + prepared: prepareMcpForm(requests[key], { name: toolName, source: serverId }), + })); + for (const { prepared } of forms) { + if ( + JSON.stringify(deepScrub(prepared.form, inventory)) !== JSON.stringify(prepared.form) + ) { + throw new McpToolCallError( + serverId, + toolName, + 'form contains private continuation material', + ); + } + } + const responses: Record = {}; + for (const { key, prepared } of forms) { + assertCurrent(); + // Providers may ignore cancellationSignal. Racing the callback still + // settles their invocation, which closes the canonical Host form. + const answer = await waitForMcpForm( + () => requestInteraction(prepared.form, { cancellationSignal: signal }), + signal!, + ); + assertCurrent(); + Object.defineProperty(responses, key, { + value: prepared.respond(answer), + enumerable: true, + configurable: true, + writable: true, + }); + } + continuation = { + ...(keys.length ? { inputResponses: responses } : {}), + ...(state === undefined ? {} : { requestState: state }), + }; + if (!keys.length) await delay(25, undefined, { signal }); + } + // The SDK's legacy compatibility schema defaults a missing content array + // before returning, but retains deferred-result compatibility fields. + // Reject every decoded marker and any future content-less result. + if ( + !Object.hasOwn(result, 'content') || + Object.hasOwn(result, 'toolResult') || + Object.hasOwn(result, 'task') || + Object.hasOwn(result, 'inputRequests') || + Object.hasOwn(result, 'requestState') + ) { + throw new McpToolCallError( + serverId, + toolName, + 'server returned an unsupported deferred tool result', + ); + } + if (!Array.isArray(result.content)) { + throw new McpToolCallError(serverId, toolName, 'server returned invalid content'); + } + if (result.isError) { + // The server writes this text; it can echo a secret it was sent. + throw new McpToolCallError( + serverId, + toolName, + scrubKnownSecrets(redactSecrets(summarizeErrorContent(result.content)), inventory), + ); + } + assertSuccessfulToolResultBudget(serverId, toolName, result, { raw: true }); + const validateOutput = preparation.value.validateOutput; + if (validateOutput) { + if (result.structuredContent === undefined) { + throw new McpToolCallError(serverId, toolName, 'server returned an invalid tool result'); + } + let validation; + try { + validation = validateOutput(result.structuredContent); + } catch (cause) { + throw new McpToolCallError(serverId, toolName, 'server returned an invalid tool result', { + cause, + }); + } + if (!validation.valid) { + throw new McpToolCallError(serverId, toolName, 'server returned an invalid tool result', { + cause: new Error(validation.errorMessage), + }); + } + } + // Success payloads cross toward the renderer and the transcript too; a + // server can embed the credential it was just sent into a result. + const published = { + content: deepScrub(result.content.map(normalizeContent), inventory), + structuredContent: deepScrub(result.structuredContent, inventory), + }; + assertSuccessfulToolResultBudget(serverId, toolName, published); + return published; } catch (error) { - // A 401 after connect means the server revoked the session, not that - // one call hiccuped: leave `connected` and the UI never offers the - // login it now needs. + refreshInventory(); + const normalized = normalizeToolCallError(serverId, toolName, error, signal); if ( isAuthRequiredError(error) && this.connections.get(serverId) === entry && entry.client === client ) { - this.markError(entry, error); + this.markError(entry, scrubbedError(error, inventory)); } - // The transport error can carry reflected request material (the body - // of a failed POST); scrub it like every other outbound message. The - // cause chain still holds the RAW transport error — the rejection - // leaves the manager (IPC, logs, telemetry), and any cause-aware - // serializer downstream would expose it — so the retained cause is an - // allowlisted copy: typed identity (name, code, status, SDK brands) - // with a scrubbed message and no deeper chain or payload fields. - const normalized = normalizeToolCallError(serverId, toolName, error, options.signal); - normalized.message = scrubKnownSecrets(normalized.message, inventory); + normalized.message = privacy?.exhausted + ? FORM_PRIVACY_EXHAUSTED + : scrubKnownSecrets(normalized.message, inventory); normalized.cause = sanitizedCause(normalized.cause, inventory); throw normalized; - } - // The SDK's legacy compatibility schema defaults a missing content array - // before returning, but retains deferred-result compatibility fields. - // Reject every decoded marker and any future content-less result. - if ( - !Object.hasOwn(result, 'content') || - Object.hasOwn(result, 'toolResult') || - Object.hasOwn(result, 'task') || - Object.hasOwn(result, 'inputRequests') || - Object.hasOwn(result, 'requestState') - ) { - throw new McpToolCallError( - serverId, - toolName, - 'server returned an unsupported deferred tool result', - ); - } - if (!Array.isArray(result.content)) { - throw new McpToolCallError(serverId, toolName, 'server returned invalid content'); - } - if (result.isError) { - // The server writes this text; it can echo a secret it was sent. - throw new McpToolCallError( - serverId, - toolName, - scrubKnownSecrets(redactSecrets(summarizeErrorContent(result.content)), inventory), - ); - } - assertSuccessfulToolResultBudget(serverId, toolName, result, { raw: true }); - const validateOutput = preparation.value.validateOutput; - if (validateOutput) { - if (result.structuredContent === undefined) { - throw new McpToolCallError(serverId, toolName, 'server returned an invalid tool result'); - } - let validation; - try { - validation = validateOutput(result.structuredContent); - } catch (cause) { - throw new McpToolCallError(serverId, toolName, 'server returned an invalid tool result', { - cause, - }); - } - if (!validation.valid) { - throw new McpToolCallError(serverId, toolName, 'server returned an invalid tool result', { - cause: new Error(validation.errorMessage), - }); + } finally { + if (controller) { + this.formCalls.delete(controller); + if (privacy) { + privacy.discardPendingStderr?.(); + privacy.pendingCalls -= 1; + } } } - // Success payloads cross toward the renderer and the transcript too; a - // server can embed the credential it was just sent into a result. - const published = { - content: deepScrub(result.content.map(normalizeContent), inventory), - structuredContent: deepScrub(result.structuredContent, inventory), - }; - assertSuccessfulToolResultBudget(serverId, toolName, published); - return published; } async test(serverId: string): Promise { @@ -919,6 +1132,14 @@ export class McpClientManager { ): Promise { let connected: OpenedMcpClient | undefined; entry.closing = false; + entry.diagnosticPrivacy = { + states: new Set(), + failure: new AbortController(), + bytes: 0, + pendingCalls: 0, + stderrSuppressed: false, + exhausted: false, + }; entry.refreshDiagnostic = undefined; entry.subscription = undefined; entry.subscriptionDiagnostic = undefined; @@ -1131,9 +1352,15 @@ export class McpClientManager { ), stderr: 'pipe', }); - attachStderrTail(transport, entry, collectConfigSecrets(entry.config), () => { - if (this.connections.get(serverId) === entry) this.emit(entry.status); - }); + attachStderrTail( + transport, + entry, + entry.diagnosticPrivacy!, + () => this.secretsFor(serverId, entry.config), + () => { + if (this.connections.get(serverId) === entry) this.emit(entry.status); + }, + ); const { client, events } = this.createClient(resolveMcpProtocolPreference(entry.config)); const isClosed = this.watchClientClose(serverId, entry, client); try { @@ -2053,6 +2280,13 @@ export class McpClientManager { entry.toolSnapshot = snapshot; this.bindingIndex = nextIndex; this.callableSnapshot = nextCallableSnapshot; + // Notify after publishing the replacement so every resumed callback sees + // the same retired binding. Unchanged refreshes preserve binding tokens. + for (const [controller, call] of this.formCalls) { + if (call.serverId === entry.status.serverId && !nextIndex.has(call.binding)) { + controller.abort(new Error('MCP tool binding is stale')); + } + } } private buildCallableSnapshot( @@ -2601,19 +2835,88 @@ export function buildStdioEnvironment( return environment; } +function exhaustConnectionDiagnosticPrivacy( + entry: Connection, + privacy: ConnectionDiagnosticPrivacy, +): void { + privacy.exhausted = true; + privacy.failure.abort(new Error(FORM_PRIVACY_EXHAUSTED)); + privacy.stderrSuppressed = true; + privacy.discardPendingStderr?.(); + entry.status = { ...entry.status, stderrTail: undefined }; +} + +function retainConnectionFormState( + entry: Connection, + privacy: ConnectionDiagnosticPrivacy, + state: string, +): boolean { + if (/[^\x20-\x7e]/u.test(state)) { + // Line splitting/formatting can remove controls, and chunk-wise UTF-8 + // decoding can split non-ASCII state bytes. Never publish this generation's + // stderr when exact raw-string replacement cannot cover those transforms. + privacy.stderrSuppressed = true; + privacy.discardPendingStderr?.(); + } + if (privacy.states.has(state)) return true; + const bytes = Buffer.byteLength(state); + if ( + privacy.states.size >= MAX_CONNECTION_FORM_STATES || + privacy.bytes + bytes > MAX_CONNECTION_FORM_STATE_BYTES + ) { + exhaustConnectionDiagnosticPrivacy(entry, privacy); + return false; + } + privacy.states.add(state); + privacy.bytes += bytes; + return true; +} + function attachStderrTail( transport: StdioClientTransport, entry: Connection, - secrets: SecretInventory, + privacy: ConnectionDiagnosticPrivacy, + secrets: () => SecretInventory, onUpdate: () => void, ): void { let pending = ''; let oversized = false; let discardingContinuation = false; let physicalSuffix = ''; + let sensitivePartialLine = false; + let sensitiveSuffix = ''; + const discardSensitive = (value: string, all: boolean): number => { + let offset = 0; + while (offset < value.length && (all || sensitivePartialLine)) { + const newline = value.indexOf('\n', offset); + const part = value.slice(offset, newline < 0 ? undefined : newline); + sensitiveSuffix = part.length >= 2 ? part.slice(-2) : `${sensitiveSuffix}${part}`.slice(-2); + if (newline < 0) { + sensitivePartialLine = true; + return value.length; + } + sensitivePartialLine = sensitiveSuffix.endsWith('\\') || sensitiveSuffix.endsWith('\\\r'); + sensitiveSuffix = ''; + offset = newline + 1; + } + return offset; + }; + const current = () => entry.diagnosticPrivacy === privacy && !entry.closing; + const suppressed = () => privacy.stderrSuppressed || privacy.pendingCalls > 0; + privacy.discardPendingStderr = () => { + if (pending.length > 0 || oversized || discardingContinuation) { + sensitivePartialLine = true; + sensitiveSuffix = physicalSuffix; + } + pending = ''; + oversized = false; + discardingContinuation = false; + physicalSuffix = ''; + }; const append = (lines: string[]) => { + if (!current() || suppressed()) return; const rendered = lines - .map((line) => formatMcpDiagnosticText(scrubKnownSecrets(line, secrets), STDERR_LINE_CHARS)) + .map((line) => formatMcpDiagnosticText(scrubKnownSecrets(line, secrets()), STDERR_LINE_CHARS)) .filter(Boolean); if (rendered.length === 0) return; const next = [...(entry.status.stderrTail ?? []), ...rendered] @@ -2661,8 +2964,14 @@ function attachStderrTail( const stream = transport.stderr; stream?.on('data', (chunk) => { const batch: string[] = []; + if (!current()) return; const value = String(chunk); - let offset = 0; + if (suppressed()) { + privacy.discardPendingStderr?.(); + discardSensitive(value, true); + return; + } + let offset = discardSensitive(value, false); for (let newline = value.indexOf('\n', offset); newline >= 0; ) { consume(value.slice(offset, newline), true, batch); offset = newline + 1; @@ -2672,6 +2981,7 @@ function attachStderrTail( append(batch); }); const flush = () => { + if (!current() || suppressed() || sensitivePartialLine) return; if (!pending && !oversized && !discardingContinuation) return; const batch: string[] = []; retain( @@ -3092,3 +3402,25 @@ class McpAbandonSupersededError extends Error { function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value); } + +/** Race even handlers that do not accept a cancellation signal, and always + * consume a late callback rejection after the invocation has been released. */ +async function waitForMcpForm(request: () => Promise, signal: AbortSignal): Promise { + signal.throwIfAborted(); + let abort!: () => void; + const cancelled = new Promise((_, reject) => { + abort = () => reject(signal.reason ?? new Error('MCP form cancelled')); + signal.addEventListener('abort', abort, { once: true }); + }); + try { + return await Promise.race([ + Promise.resolve().then(() => { + signal.throwIfAborted(); + return request(); + }), + cancelled, + ]); + } finally { + signal.removeEventListener('abort', abort); + } +} diff --git a/scripts/release-cli-file-policy.test.mjs b/scripts/release-cli-file-policy.test.mjs index 84e7e8e371..0cbf0c401f 100644 --- a/scripts/release-cli-file-policy.test.mjs +++ b/scripts/release-cli-file-policy.test.mjs @@ -56,7 +56,7 @@ describe('CLI release file policy', () => { const repoRoot = resolve(import.meta.dirname, '..'); for (const [directory, omitted] of Object.entries({ core: ['./test-only/async-primitives'], - mcp: ['./test-only/stdio-server'], + mcp: ['./test-only/stdio-server', './test-only/form-server'], 'runtime-host': [ './test-only/client-capability-host', './test-only/execution-candidate-e2e-main',