From b6e270ee4856ae6adde66789e42c068eb489100f Mon Sep 17 00:00:00 2001 From: Michal Date: Tue, 25 Aug 2026 23:41:09 +0100 Subject: [PATCH] feat(mcplocal): every request answers, even when the pipeline wedges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The LLM budgets bound the failure we actually hit. This is the backstop for the ones they cannot see: a wedged plugin hook, a virtual-tool handler, an upstream path without its own timeout. transport.onmessage IS the whole request pipeline, and it writes to a socket that has already been hijacked from Fastify. The SDK does not await it, so a throw became an unhandled rejection and a hang became silence -- in both cases the client got no response at all and waited out its own timeout. onmessage is now structured so that reaching transport.send() is unconditional: the route call is raced against a deadline, and the notification flush and the send each carry their own catch, so a failure while flushing can no longer cost the client its response. transport.onerror was never assigned, so SDK-level transport errors were swallowed entirely. It is now. Shape of the answer is deliberate. A tools/call comes back as a SUCCESSFUL result with isError and readable text naming the trace code -- the same shape router.ts already uses for an expired _resultId, because a transport error tends to surface as a hard failure while a tool error is something the model reads and acts on. It also says the upstream may still be running, which is true and matters. Other methods get a JSON-RPC error (-32001). MCPLOCAL_TOOLCALL_DEADLINE_MS defaults to TOOLCALL_TIMEOUT_MS + 30s rather than 120s. The watchdog arms earlier in the request than mcpd's fetch does, so at equal values it would always fire first, masking mcpd's specific UpstreamTimeoutError with a generic message and cutting off pagination and the rest of the post-processing. There is a test asserting the two stay ordered. The pause queue is exempt, but not bypassed. It blocks until a human operator releases, edits or drops a response, so the deadline SUSPENDS -- clock stopped, then re-armed with whatever was left. A test asserts the re-arm, because an exemption that forgot to re-arm would silently make every paused request immortal. End-to-end: an upstream that never settles now yields a response in under a second carrying ⚠, the deadline, and `mcpctl trace `. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GqMidYEGUJG5fxeoTELBu2 --- src/mcplocal/src/http/deadline.ts | 75 ++++++++++++ src/mcplocal/src/http/project-mcp-endpoint.ts | 113 +++++++++++++++-- src/mcplocal/src/request-context.ts | 7 ++ src/mcplocal/src/router.ts | 11 +- .../tests/toolcall-deadline-e2e.test.ts | 115 ++++++++++++++++++ src/mcplocal/tests/toolcall-deadline.test.ts | 74 +++++++++++ 6 files changed, 380 insertions(+), 15 deletions(-) create mode 100644 src/mcplocal/src/http/deadline.ts create mode 100644 src/mcplocal/tests/toolcall-deadline-e2e.test.ts create mode 100644 src/mcplocal/tests/toolcall-deadline.test.ts diff --git a/src/mcplocal/src/http/deadline.ts b/src/mcplocal/src/http/deadline.ts new file mode 100644 index 0000000..44a4b80 --- /dev/null +++ b/src/mcplocal/src/http/deadline.ts @@ -0,0 +1,75 @@ +/** + * Outer bound on one client request. + * + * The LLM budgets (proxymodel/llm-adapter.ts, proxymodel/stage-budget.ts) bound + * the failure mode we actually hit in production. This is the backstop for the + * ones they cannot see: a wedged plugin hook, a virtual-tool handler, an + * upstream path without its own timeout. `transport.onmessage` writes to an + * already-hijacked socket, so if it never completes the client gets silence and + * waits out its own timeout — 1800s in Claude Code. + * + * MUST exceed McpdClient's TOOLCALL_TIMEOUT_MS. The watchdog arms earlier in + * the request than the mcpd fetch does, so at equal values it would always fire + * first and mask mcpd's specific UpstreamTimeoutError with a generic deadline + * message — and cut off pagination and the rest of the post-processing besides. + */ +import { TOOLCALL_TIMEOUT_MS } from './mcpd-client.js'; + +export const TOOLCALL_DEADLINE_MS = + Number(process.env['MCPLOCAL_TOOLCALL_DEADLINE_MS']) || (TOOLCALL_TIMEOUT_MS + 30_000); + +export class DeadlineExceededError extends Error { + constructor(public readonly method: string, public readonly ms: number) { + super(`${method} exceeded the ${String(ms)}ms mcplocal deadline`); + this.name = 'DeadlineExceededError'; + } +} + +export interface RequestDeadline { + /** Rejects with DeadlineExceededError when the budget runs out. */ + readonly expiry: Promise; + /** Run `fn` with the clock stopped, then re-arm with what was left. */ + suspended(fn: () => Promise): Promise; + dispose(): void; +} + +export function createRequestDeadline(method: string, totalMs: number): RequestDeadline { + let remaining = totalMs; + let startedAt = Date.now(); + let timer: ReturnType | undefined; + let reject!: (err: Error) => void; + let settled = false; + + const expiry = new Promise((_, rej) => { reject = rej; }); + // Nothing else may observe this rejection until it is raced, and an unraced + // rejection would be an unhandled rejection at process level. + expiry.catch(() => { /* raced by the caller */ }); + + const arm = (ms: number): void => { + startedAt = Date.now(); + timer = setTimeout(() => { + settled = true; + reject(new DeadlineExceededError(method, totalMs)); + }, ms); + if (typeof timer.unref === 'function') timer.unref(); + }; + arm(remaining); + + return { + expiry, + async suspended(fn: () => Promise): Promise { + if (settled) return fn(); + if (timer !== undefined) clearTimeout(timer); + remaining = Math.max(0, remaining - (Date.now() - startedAt)); + try { + return await fn(); + } finally { + if (!settled) arm(remaining); + } + }, + dispose(): void { + settled = true; + if (timer !== undefined) clearTimeout(timer); + }, + }; +} diff --git a/src/mcplocal/src/http/project-mcp-endpoint.ts b/src/mcplocal/src/http/project-mcp-endpoint.ts index 8b298de..d6e56b8 100644 --- a/src/mcplocal/src/http/project-mcp-endpoint.ts +++ b/src/mcplocal/src/http/project-mcp-endpoint.ts @@ -32,6 +32,7 @@ import { newTraceCode } from '../util/trace-code.js'; import { runInRequestScope } from '../request-context.js'; import { adoptSession } from './session-adopt.js'; import { degradationNotice } from '../util/degrade.js'; +import { createRequestDeadline, DeadlineExceededError, TOOLCALL_DEADLINE_MS } from './deadline.js'; interface ProjectCacheEntry { router: McpRouter; @@ -246,6 +247,53 @@ export function registerProjectMcpEndpoint(app: FastifyInstance, mcpdClient: Mcp return response; } + /** + * Turn a deadline or an unexpected throw into something the client can read. + * + * A tools/call becomes a SUCCESSFUL result with isError — a transport error + * tends to surface to the user as a hard failure, whereas a tool error is + * text the model reads and acts on (the same shape router.ts already uses for + * an expired _resultId). Everything else gets a JSON-RPC error. + */ + function failureResponse( + requestId: string | number, + method: string | undefined, + err: unknown, + traceCode: string, + ): unknown { + const timedOut = err instanceof DeadlineExceededError; + const detail = timedOut + ? `exceeded the ${String(TOOLCALL_DEADLINE_MS)}ms mcplocal deadline and was abandoned` + : `failed: ${err instanceof Error ? err.message : String(err)}`; + console.error(`[mcp] ${method ?? 'request'} ${detail} (trace ${traceCode})`); + + if (method === 'tools/call') { + return { + jsonrpc: '2.0', + id: requestId, + result: { + content: [{ + type: 'text', + text: + `⚠ This tool call ${detail} (trace ${traceCode}).\n` + + 'The upstream may still be running — do not assume it did nothing. ' + + 'Retry with narrower arguments, or run ' + + `\`mcpctl trace ${traceCode}\` to see which stage consumed the time.`, + }], + isError: true, + }, + }; + } + return { + jsonrpc: '2.0', + id: requestId, + error: { + code: timedOut ? -32001 : -32603, + message: `${method ?? 'request'} ${detail} (trace ${traceCode})`, + }, + }; + } + app.post<{ Params: { projectName: string } }>('/projects/:projectName/mcp', async (request, reply) => { const { projectName } = request.params; const sessionId = request.headers['mcp-session-id'] as string | undefined; @@ -396,15 +444,35 @@ export function registerProjectMcpEndpoint(app: FastifyInstance, mcpdClient: Mcp // Everything downstream — plugins, stages, the pipeline executor — is // built per SESSION, not per request, so the trace code reaches them // through the async scope rather than a parameter on every signature. - const response = await runInRequestScope( - { correlationId, sessionId: sid, projectName, method: method ?? 'unknown' }, - () => routeWithWireNames( - codec, - (req) => router.route(req, ctx), - message as unknown as JsonRpcRequest, - ), - ); + // + // The deadline is the backstop for hangs the LLM budgets cannot see: a + // wedged plugin hook, a virtual-tool handler, an upstream without its + // own timeout. Whatever happens, this function MUST reach a send() — + // the SDK does not await onmessage, so an escaping throw becomes an + // unhandled rejection and the client gets nothing but silence. + const deadline = createRequestDeadline(method ?? 'request', TOOLCALL_DEADLINE_MS); + let response: unknown; + try { + response = await Promise.race([ + runInRequestScope( + { correlationId, sessionId: sid, projectName, method: method ?? 'unknown', deadline }, + () => routeWithWireNames( + codec, + (req) => router.route(req, ctx), + message as unknown as JsonRpcRequest, + ), + ), + deadline.expiry, + ]); + } catch (err) { + response = failureResponse(requestId, method, err, correlationId); + } finally { + deadline.dispose(); + } + // Guaranteed-send tail. Everything from here on is best-effort: a throw + // while flushing notifications must not cost the client its response. + try { // Forward queued notifications BEFORE the response — the response send // closes the POST SSE stream, so notifications must go first. // relatedRequestId routes them onto the same SSE stream as the response. @@ -434,13 +502,34 @@ export function registerProjectMcpEndpoint(app: FastifyInstance, mcpdClient: Mcp correlationId, }); - requestCorrelations.delete(requestId); - await transport.send( - maybeAnnotateRecreation(response, sid, method) as unknown as JSONRPCMessage, - ); + } catch (tailErr) { + console.error( + `[mcp] notification flush failed for trace ${correlationId}: ` + + `${tailErr instanceof Error ? tailErr.message : String(tailErr)}`, + ); + } + + try { + await transport.send( + maybeAnnotateRecreation(response, sid, method) as unknown as JSONRPCMessage, + ); + } catch (sendErr) { + console.error( + `[mcp] failed to deliver response for trace ${correlationId} (${method ?? 'unknown'}): ` + + `${sendErr instanceof Error ? sendErr.message : String(sendErr)}`, + ); + } finally { + requestCorrelations.delete(requestId); + } } }; + // Never assigned before this change, so an SDK-level transport error was + // swallowed entirely. + transport.onerror = (err: Error) => { + console.error(`[mcp] transport error (${projectName}): ${err.message}`); + }; + transport.onclose = () => { const id = transport.sessionId; if (id) { diff --git a/src/mcplocal/src/request-context.ts b/src/mcplocal/src/request-context.ts index 25f8318..f0e225a 100644 --- a/src/mcplocal/src/request-context.ts +++ b/src/mcplocal/src/request-context.ts @@ -12,6 +12,7 @@ * requests are in flight on the same session. */ import { AsyncLocalStorage } from 'node:async_hooks'; +import type { RequestDeadline } from './http/deadline.js'; export interface RequestScope { /** The trace code; also the audit correlationId. */ @@ -19,6 +20,12 @@ export interface RequestScope { sessionId: string; projectName: string; method: string; + /** + * The request's deadline, so deliberately-unbounded waits (the pause queue, + * which blocks until a human operator releases a response) can stop its clock + * instead of being killed by it. + */ + deadline?: RequestDeadline | undefined; } const storage = new AsyncLocalStorage(); diff --git a/src/mcplocal/src/router.ts b/src/mcplocal/src/router.ts index 06fbbf1..02eba13 100644 --- a/src/mcplocal/src/router.ts +++ b/src/mcplocal/src/router.ts @@ -6,7 +6,7 @@ import type { PromptIndexEntry } from './gate/tag-matcher.js'; import { LinkResolver } from './services/link-resolver.js'; import type { LLMProvider, CacheProvider, Section } from './proxymodel/types.js'; import { executePipeline } from './proxymodel/executor.js'; -import { currentCorrelationId } from './request-context.js'; +import { currentCorrelationId, currentScope } from './request-context.js'; import { getProxyModel } from './proxymodel/loader.js'; import type { ProxyModelPlugin, PluginSessionContext } from './proxymodel/plugin.js'; import { PluginContextImpl, type PluginContextDeps } from './proxymodel/plugin-context.js'; @@ -183,9 +183,12 @@ export class McpRouter { ...(currentCorrelationId() !== undefined ? { correlationId: currentCorrelationId()! } : {}), }); - // Pause queue: if paused, hold the result until released/edited/dropped + // Pause queue: if paused, hold the result until released/edited/dropped. + // This wait is deliberately unbounded — it ends when a human operator + // acts — so the request deadline stops its clock rather than killing + // it, and re-arms with whatever was left once the operator releases. if (pauseQueue.paused) { - const pausedContent = await pauseQueue.enqueue({ + const enqueue = (): Promise => pauseQueue.enqueue({ sessionId, projectName: this.projectName ?? 'unknown', contentType, @@ -193,6 +196,8 @@ export class McpRouter { original: content, transformed: result.content, }); + const deadline = currentScope()?.deadline; + const pausedContent = deadline ? await deadline.suspended(enqueue) : await enqueue(); return { ...result, content: pausedContent }; } diff --git a/src/mcplocal/tests/toolcall-deadline-e2e.test.ts b/src/mcplocal/tests/toolcall-deadline-e2e.test.ts new file mode 100644 index 0000000..f69bc2c --- /dev/null +++ b/src/mcplocal/tests/toolcall-deadline-e2e.test.ts @@ -0,0 +1,115 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import Fastify from 'fastify'; +import { registerProjectMcpEndpoint } from '../src/http/project-mcp-endpoint.js'; +import type { McpRouter } from '../src/router.js'; +import type { JsonRpcRequest, JsonRpcResponse } from '../src/types.js'; + +/** + * The guarantee this whole change exists for: a tool call whose upstream never + * settles must still produce a response. Before the watchdog, onmessage simply + * never reached transport.send(), the hijacked socket stayed silent, and the + * client waited out its own 1800s timeout — which is precisely what happened + * in production three times. + */ + +vi.mock('../src/discovery.js', () => ({ + refreshProjectUpstreams: vi.fn(async (router: McpRouter) => { + router.addUpstream({ + name: 'blackhole', + send: async (req: JsonRpcRequest): Promise => { + if (req.method === 'tools/list') { + return { + jsonrpc: '2.0', + id: req.id, + result: { tools: [{ name: 'wedge', description: 'never returns', inputSchema: { type: 'object' } }] }, + }; + } + // A call that never settles and never errors. + return new Promise(() => { /* the hang */ }); + }, + close: async () => { /* noop */ }, + isAlive: () => true, + }); + return ['blackhole']; + }), + fetchProjectLlmConfig: vi.fn(async () => ({ gated: false, llmProvider: 'none' })), +})); + +vi.mock('../src/http/config.js', async () => { + const actual = await vi.importActual('../src/http/config.js'); + return { ...actual, loadProjectLlmOverride: vi.fn(() => undefined) }; +}); + +function mockMcpdClient(): Record { + const client: Record = { + baseUrl: 'http://test:3100', token: 't', + get: vi.fn(async () => []), post: vi.fn(async () => ({})), + put: vi.fn(), delete: vi.fn(), + forward: vi.fn(async () => ({ status: 200, body: [] })), + withHeaders: vi.fn(), withToken: vi.fn(), withTimeout: vi.fn(), + }; + for (const k of ['withHeaders', 'withToken', 'withTimeout']) { + (client[k] as ReturnType).mockReturnValue(client); + } + return client; +} + +function parseSse(body: string): JsonRpcResponse { + const line = body.split('\n').find((l) => l.startsWith('data: ')); + if (!line) throw new Error(`no SSE data line in: ${body}`); + return JSON.parse(line.slice('data: '.length)) as JsonRpcResponse; +} + +beforeEach(() => { vi.stubEnv('MCPLOCAL_TOOLCALL_DEADLINE_MS', '300'); vi.resetModules(); }); + +describe('tool-call watchdog, end to end', () => { + it('answers a wedged tool call instead of leaving the client hanging', async () => { + vi.spyOn(console, 'error').mockImplementation(() => { /* quiet */ }); + const { registerProjectMcpEndpoint: register } = + await import('../src/http/project-mcp-endpoint.js'); + + const app = Fastify(); + register(app, mockMcpdClient() as never); + await app.ready(); + try { + const headers = { 'content-type': 'application/json', accept: 'application/json, text/event-stream' }; + const init = await app.inject({ + method: 'POST', url: '/projects/wedge-test/mcp', headers, + payload: { jsonrpc: '2.0', id: 1, method: 'initialize', params: { protocolVersion: '2024-11-05', capabilities: {}, clientInfo: { name: 't', version: '0' } } }, + }); + const sessionId = init.headers['mcp-session-id'] as string; + expect(sessionId).toBeTruthy(); + + // tools/list first: the wire-name codec learns the mapping there. + await app.inject({ + method: 'POST', url: '/projects/wedge-test/mcp', + headers: { ...headers, 'mcp-session-id': sessionId }, + payload: { jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }, + }); + + const start = Date.now(); + const res = await app.inject({ + method: 'POST', url: '/projects/wedge-test/mcp', + headers: { ...headers, 'mcp-session-id': sessionId }, + payload: { jsonrpc: '2.0', id: 3, method: 'tools/call', params: { name: 'blackhole_wedge', arguments: {} } }, + }); + const elapsed = Date.now() - start; + + expect(elapsed).toBeLessThan(5000); + const body = parseSse(res.body); + expect(body.id).toBe(3); + + // A tools/call comes back as a RESULT with isError, not a transport + // error: it is text the model reads and acts on. + const result = body.result as { content?: Array<{ text?: string }>; isError?: boolean }; + expect(result.isError).toBe(true); + const text = result.content?.[0]?.text ?? ''; + expect(text).toContain('⚠'); + expect(text).toContain('mcplocal deadline'); + // The trace code must be present and actionable. + expect(text).toMatch(/mcpctl trace [0-9ABCDEFGHJKMNPQRSTVWXYZ]{8}/); + } finally { + await app.close(); + } + }); +}); diff --git a/src/mcplocal/tests/toolcall-deadline.test.ts b/src/mcplocal/tests/toolcall-deadline.test.ts new file mode 100644 index 0000000..b06f8a2 --- /dev/null +++ b/src/mcplocal/tests/toolcall-deadline.test.ts @@ -0,0 +1,74 @@ +import { describe, it, expect, vi, afterEach } from 'vitest'; +import { + createRequestDeadline, + DeadlineExceededError, + TOOLCALL_DEADLINE_MS, +} from '../src/http/deadline.js'; +import { TOOLCALL_TIMEOUT_MS } from '../src/http/mcpd-client.js'; + +afterEach(() => { vi.restoreAllMocks(); }); + +describe('TOOLCALL_DEADLINE_MS', () => { + it('exceeds mcpd\'s own tool-call timeout', () => { + // The watchdog arms earlier in the request than mcpd's fetch does. At equal + // values it would always fire first, masking mcpd's specific + // UpstreamTimeoutError with a generic deadline message and cutting off + // pagination and the rest of the post-processing. + expect(TOOLCALL_DEADLINE_MS).toBeGreaterThan(TOOLCALL_TIMEOUT_MS); + }); + + it('stays well below a typical client timeout so our error arrives first', () => { + expect(TOOLCALL_DEADLINE_MS).toBeLessThan(600_000); + }); +}); + +describe('createRequestDeadline', () => { + it('rejects with DeadlineExceededError when the budget runs out', async () => { + const d = createRequestDeadline('tools/call', 40); + await expect(Promise.race([ + new Promise(() => { /* the hang we are protecting against */ }), + d.expiry, + ])).rejects.toBeInstanceOf(DeadlineExceededError); + d.dispose(); + }); + + it('does not fire once disposed', async () => { + const d = createRequestDeadline('tools/call', 30); + d.dispose(); + const outcome = await Promise.race([ + d.expiry.then(() => 'fired', () => 'fired'), + new Promise((r) => setTimeout(() => { r('quiet'); }, 120)), + ]); + expect(outcome).toBe('quiet'); + }); + + it('stops the clock while suspended, so an operator pause cannot time out', async () => { + const d = createRequestDeadline('tools/call', 100); + + // A pause queue wait far longer than the whole deadline. + const held = await d.suspended(async () => { + await new Promise((r) => setTimeout(r, 250)); + return 'released'; + }); + expect(held).toBe('released'); + + // Still alive afterwards, with roughly the original budget left. + const outcome = await Promise.race([ + d.expiry.then(() => 'fired', () => 'fired'), + new Promise((r) => setTimeout(() => { r('still-running'); }, 40)), + ]); + expect(outcome).toBe('still-running'); + d.dispose(); + }); + + it('re-arms after suspension rather than becoming immortal', async () => { + const d = createRequestDeadline('tools/call', 60); + await d.suspended(async () => { await new Promise((r) => setTimeout(r, 100)); }); + // The remaining budget must still expire — suspending is not a bypass. + await expect(Promise.race([ + new Promise(() => { /* hang */ }), + d.expiry, + ])).rejects.toBeInstanceOf(DeadlineExceededError); + d.dispose(); + }); +});