feat(mcplocal): every request answers, even when the pipeline wedges

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 <code>`.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GqMidYEGUJG5fxeoTELBu2

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<never>;
+  /** Run `fn` with the clock stopped, then re-arm with what was left. */
+  suspended<T>(fn: () => Promise<T>): Promise<T>;
+  dispose(): void;
+}
+
+export function createRequestDeadline(method: string, totalMs: number): RequestDeadline {
+  let remaining = totalMs;
+  let startedAt = Date.now();
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let reject!: (err: Error) => void;
+  let settled = false;
+
+  const expiry = new Promise<never>((_, 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<T>(fn: () => Promise<T>): Promise<T> {
+      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);
+    },
+  };
+}

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) {

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<RequestScope>();

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<string> => 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 };
         }
 

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<JsonRpcResponse> => {
+        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<JsonRpcResponse>(() => { /* 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<typeof import('../src/http/config.js')>('../src/http/config.js');
+  return { ...actual, loadProjectLlmOverride: vi.fn(() => undefined) };
+});
+
+function mockMcpdClient(): Record<string, unknown> {
+  const client: Record<string, unknown> = {
+    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<typeof vi.fn>).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();
+    }
+  });
+});

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<never>(() => { /* 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<never>(() => { /* hang */ }),
+      d.expiry,
+    ])).rejects.toBeInstanceOf(DeadlineExceededError);
+    d.dispose();
+  });
+});