diff --git a/src/McpServer/autopilot-monitor-mcp/src/__tests__/telemetry.test.ts b/src/McpServer/autopilot-monitor-mcp/src/__tests__/telemetry.test.ts index 83fea22a..6cc1054f 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/__tests__/telemetry.test.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/__tests__/telemetry.test.ts @@ -150,3 +150,70 @@ describe('logToolCallRejection', () => { expect((line.message as string).length).toBeLessThan(530); }); }); + +describe('attachToolCallRejectionSniffer', () => { + const ERROR_ENVELOPE = JSON.stringify({ + jsonrpc: '2.0', + id: 1, + error: { code: -32602, message: 'Input validation error: Invalid arguments for tool get_session: sessionId must be a UUID' }, + }); + + function fakeRes() { + const calls: string[] = []; + return { + calls, + write: vi.fn((..._args: never[]) => { calls.push('write'); return true; }), + end: vi.fn((..._args: never[]) => { calls.push('end'); return undefined as unknown; }), + }; + } + + it('catches the Hono streaming path: Uint8Array chunks via write, then a bare end()', async () => { + const { attachToolCallRejectionSniffer } = await loadTelemetry(true); + const spy = vi.spyOn(console, 'error').mockImplementation(() => {}); + const res = fakeRes(); + attachToolCallRejectionSniffer('get_session', res); + + const bytes = new TextEncoder().encode(ERROR_ENVELOPE); + // Split mid-envelope to prove reassembly across chunks works. + res.write(bytes.slice(0, 40) as never); + res.write(bytes.slice(40) as never); + res.end(); + + const line = lastLoggedJson(spy); + expect(line.type).toBe('tool_call_rejected'); + expect(line.tool).toBe('get_session'); + expect(line.errorCode).toBe(-32602); + expect(line.message).toContain('sessionId must be a UUID'); + // The response itself went through untouched. + expect(res.calls).toEqual(['write', 'write', 'end']); + }); + + it('catches the direct path: whole envelope as a single end(string)', async () => { + const { attachToolCallRejectionSniffer } = await loadTelemetry(true); + const spy = vi.spyOn(console, 'error').mockImplementation(() => {}); + const res = fakeRes(); + attachToolCallRejectionSniffer('query_table', res); + res.end(ERROR_ENVELOPE as never); + expect(lastLoggedJson(spy).tool).toBe('query_table'); + }); + + it('stays silent for successful results and stops buffering past the cap', async () => { + const { attachToolCallRejectionSniffer } = await loadTelemetry(true); + const spy = vi.spyOn(console, 'error').mockImplementation(() => {}); + const res = fakeRes(); + attachToolCallRejectionSniffer('search_sessions', res); + // A large successful tool result (which even CONTAINS the word "error" in data). + const big = JSON.stringify({ jsonrpc: '2.0', id: 1, result: { content: [{ type: 'text', text: '"error" '.repeat(3000) }] } }); + for (let i = 0; i < big.length; i += 1000) res.write(big.slice(i, i + 1000) as never); + res.end(); + expect(spy).not.toHaveBeenCalled(); + }); + + it('does not wrap the response at all when logging is disabled', async () => { + const { attachToolCallRejectionSniffer } = await loadTelemetry(false); + const res = fakeRes(); + const originalEnd = res.end; + attachToolCallRejectionSniffer('get_session', res); + expect(res.end).toBe(originalEnd); + }); +}); diff --git a/src/McpServer/autopilot-monitor-mcp/src/index.ts b/src/McpServer/autopilot-monitor-mcp/src/index.ts index ed480e58..0d4afc9a 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/index.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/index.ts @@ -18,7 +18,7 @@ import { buildEventTypeSearchDocs } from './resource-catalog.js'; import { createOAuthRouter } from './oauth.js'; import { accessGuard } from './access-guard.js'; import { hasGlobalScope, isGlobalAdmin, isDelegated, getDelegatedTenantIds, getHomeTenantId } from './client.js'; -import { toolLoggingEnabled, logToolCallRejection } from './telemetry.js'; +import { toolLoggingEnabled, attachToolCallRejectionSniffer } from './telemetry.js'; import { API_BASE_URL } from './config.js'; const __dirname = dirname(fileURLToPath(import.meta.url)); @@ -373,31 +373,15 @@ app.all('/mcp', async (req, res) => { return; } - // Sniff JSON-RPC error envelopes on tool calls (Zod argument rejection, + // Sniff the JSON-RPC error envelope on tool calls (Zod argument rejection, // unknown tool name): the SDK answers these BEFORE any tool handler runs, - // so wrapping the response write is the only place they are observable — - // and they are the strongest "the schema/description confuses the model" - // signal. Stateless mode guarantees one JSON-RPC request per POST and - // enableJsonResponse yields a single buffered body, so inspecting the - // res.end chunk sees the whole envelope. Error envelopes are tiny; the - // length guard skips parsing large (i.e. successful) tool results. + // so observing the response write is the only place they are visible — and + // they are the strongest "the schema/description confuses the model" signal. + // Stateless mode guarantees one JSON-RPC request per POST, so the tool name + // from the request body is unambiguous. if (toolLoggingEnabled && req.body?.method === 'tools/call') { const toolName = typeof req.body?.params?.name === 'string' ? req.body.params.name : 'unknown'; - const originalEnd = res.end.bind(res); - res.end = ((chunk?: unknown, ...rest: unknown[]) => { - try { - const text = typeof chunk === 'string' ? chunk : Buffer.isBuffer(chunk) ? chunk.toString('utf8') : undefined; - if (text && text.length < 10_000 && text.includes('"error"')) { - const parsed = JSON.parse(text) as { error?: { code?: number; message?: unknown } }; - if (parsed?.error?.code !== undefined) { - logToolCallRejection(toolName, parsed.error.code, String(parsed.error.message ?? '')); - } - } - } catch { - // Sniffing must never break the response. - } - return originalEnd(chunk as never, ...(rest as never[])); - }) as typeof res.end; + attachToolCallRejectionSniffer(toolName, res); } const transport = new StreamableHTTPServerTransport({ diff --git a/src/McpServer/autopilot-monitor-mcp/src/telemetry.ts b/src/McpServer/autopilot-monitor-mcp/src/telemetry.ts index 0015d0c0..4fa7e100 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/telemetry.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/telemetry.ts @@ -142,3 +142,66 @@ export function logToolCallRejection(toolName: string, errorCode: number, messag // Telemetry must never break a tool response. } } + +// Error envelopes are tiny; anything larger is a successful tool result and is +// skipped instead of buffered. +const SNIFF_BUFFER_CAP = 10_000; + +interface SniffableResponse { + write: (...args: never[]) => boolean; + end: (...args: never[]) => unknown; +} + +/** + * Observe the JSON-RPC response of a tools/call POST for an error envelope and + * log it via logToolCallRejection. The SDK (1.30+) bridges the web-standard + * Response to the Node res through Hono's getRequestListener, which depending + * on the body type either calls res.end(string | Uint8Array) directly or pipes + * a ReadableStream through res.write chunks followed by a bare res.end() — so + * BOTH must be captured. Buffering stops at SNIFF_BUFFER_CAP (real rejections + * are a few hundred bytes). Never throws into the response path. + */ +export function attachToolCallRejectionSniffer(toolName: string, res: SniffableResponse): void { + if (!toolLoggingEnabled) return; + + let captured = ''; + let overflowed = false; + const capture = (chunk: unknown): void => { + if (overflowed || chunk == null) return; + let text: string | undefined; + if (typeof chunk === 'string') text = chunk; + // Covers Node Buffers too (Buffer extends Uint8Array). + else if (chunk instanceof Uint8Array) text = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength).toString('utf8'); + if (text === undefined) return; + captured += text; + if (captured.length > SNIFF_BUFFER_CAP) { + overflowed = true; + captured = ''; + } + }; + + const originalWrite = res.write.bind(res); + const originalEnd = res.end.bind(res); + res.write = ((chunk: unknown, ...rest: unknown[]) => { + try { + capture(chunk); + } catch { + // Sniffing must never break the response. + } + return originalWrite(chunk as never, ...(rest as never[])); + }) as typeof res.write; + res.end = ((chunk?: unknown, ...rest: unknown[]) => { + try { + capture(chunk); + if (!overflowed && captured.includes('"error"')) { + const parsed = JSON.parse(captured) as { error?: { code?: number; message?: unknown } }; + if (parsed?.error?.code !== undefined) { + logToolCallRejection(toolName, parsed.error.code, String(parsed.error.message ?? '')); + } + } + } catch { + // Sniffing must never break the response. + } + return originalEnd(chunk as never, ...(rest as never[])); + }) as typeof res.end; +}