fix(mcp): bound panel requests, request bodies and cancelled calls

Panels on one MCP server share six requests in flight; further requests
wait their turn (up to 256 waiting) instead of failing, and a waiting
request leaves the queue when its caller cancels. Held responses belong to
their own MCP server rather than to the process. /mcp request bodies must
arrive within 30 seconds. Over stdio, a client's notifications/cancelled
aborts the named request, which then gets no response.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

server/mcp/plugin.js

@@ -14,6 +14,8 @@ import { createLocalMcpServer } from './server.js';
 const LOOPBACK_HOSTS = new Set(['localhost', '127.0.0.1', '[::1]']);
 const LOOPBACK_ADDRESSES = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']);
 const MAX_BODY_BYTES = 1024 * 1024;
+/** How long a client may take to send its request body. */
+const BODY_TIMEOUT_MS = 30 * 1000;
 
 /**
  * Whether a request may reach the local MCP server. `localPort` is the port
@@ -56,7 +58,10 @@ export function isLocalMcpRequest({
 }
 
 /** Vite plugin that mounts the local MCP server at /mcp. */
-export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) {
+export function localMcpPlugin({
+  createServer = createLocalMcpServer,
+  bodyTimeoutMs = BODY_TIMEOUT_MS,
+} = {}) {
   const handlers = new Map();
   const handlerFor = (apiBase) => {
     if (!handlers.has(apiBase))
@@ -94,7 +99,10 @@ export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) {
       };
       res.on('close', onClose);
       try {
-        const body = req.method === 'POST' ? await readBody(req) : undefined;
+        const body =
+          req.method === 'POST'
+            ? await readBody(req, bodyTimeoutMs)
+            : undefined;
         const request = new Request(`http://${host}/mcp`, {
           method: req.method,
           headers: Object.entries(req.headers).flatMap(([name, value]) =>
@@ -109,13 +117,18 @@ export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) {
       } catch (error) {
         if (disconnect.signal.aborted) return;
         if (res.headersSent) return res.destroy();
-        res.writeHead(error?.status === 413 ? 413 : 400, {
+        res.writeHead([408, 413].includes(error?.status) ? error.status : 400, {
           'Content-Type': 'application/json',
           'Cache-Control': 'no-store',
         });
         res.end(
           JSON.stringify({
-            error: error?.status === 413 ? 'Request too large' : 'Bad request',
+            error:
+              error?.status === 413
+                ? 'Request too large'
+                : error?.status === 408
+                  ? 'Request timed out'
+                  : 'Bad request',
           }),
         );
       } finally {
@@ -130,11 +143,21 @@ export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) {
   };
 }
 
-function readBody(req) {
+function readBody(req, timeoutMs = BODY_TIMEOUT_MS) {
   return new Promise((resolve, reject) => {
     const chunks = [];
     let size = 0;
     let tooLarge = false;
+    // A client that stops sending must not hold the request open.
+    const timer = setTimeout(
+      () =>
+        reject(Object.assign(new Error('Request timed out'), { status: 408 })),
+      timeoutMs,
+    );
+    const fail = (error) => {
+      clearTimeout(timer);
+      reject(error);
+    };
     req.on('data', (chunk) => {
       size += chunk.length;
       // Keep draining without storing, so the 413 response can still be sent.
@@ -144,9 +167,12 @@ function readBody(req) {
       }
       tooLarge = true;
       chunks.length = 0;
-      reject(Object.assign(new Error('Request too large'), { status: 413 }));
+      fail(Object.assign(new Error('Request too large'), { status: 413 }));
     });
-    req.on('end', () => resolve(Buffer.concat(chunks)));
-    req.on('error', reject);
+    req.on('end', () => {
+      clearTimeout(timer);
+      resolve(Buffer.concat(chunks));
+    });
+    req.on('error', fail);
   });
 }

server/mcp/stdio.js

@@ -30,13 +30,18 @@ export function describeRequest(message) {
   return typeof subject === 'string' ? `${method} ${subject}` : method;
 }
 
-/** Serve newline-delimited JSON-RPC until `input` ends. */
+/**
+ * Serve newline-delimited JSON-RPC until `input` ends. A client's
+ * `notifications/cancelled` aborts the named request, which then gets no
+ * response, as the protocol asks.
+ */
 export async function serveStdio(server, { input, output, log = () => {} }) {
   const lines = createInterface({ input, crlfDelay: Infinity });
   const pending = new Set();
+  const running = new Map();
   for await (const line of lines) {
     if (!line.trim()) continue;
-    const work = respond(server, line, log).then((response) => {
+    const work = respond(server, line, log, running).then((response) => {
       if (response) output.write(`${JSON.stringify(response)}\n`);
     });
     pending.add(work);
@@ -45,16 +50,30 @@ export async function serveStdio(server, { input, output, log = () => {} }) {
   await Promise.all(pending);
 }
 
-async function respond(server, line, log) {
+async function respond(server, line, log, running) {
   let message;
   try {
     message = JSON.parse(line);
   } catch {
     return parseErrorResponse();
   }
+  if (message?.method === 'notifications/cancelled') {
+    running.get(message.params?.requestId)?.abort();
+    return null;
+  }
   const described = describeRequest(message);
   if (described) log(described);
-  const response = await server.handle(message);
+  const id = message?.id;
+  const cancellable = id !== undefined && id !== null && !running.has(id);
+  const controller = new AbortController();
+  if (cancellable) running.set(id, controller);
+  let response;
+  try {
+    response = await server.handle(message, { signal: controller.signal });
+  } finally {
+    if (cancellable) running.delete(id);
+  }
+  if (controller.signal.aborted) return null;
   // A failed tool call answers the client, not the log; say why here too.
   if (response?.result?.isError)
     log(`   ${described} failed: ${response.result.content?.[0]?.text}`);

src/tools/localRoute.test.mjs

@@ -195,3 +195,44 @@ test('a client that disconnects cancels its tool call', async (t) => {
   await aborted.promise;
   assert.equal(seen.aborted, true);
 });
+
+test('a client that stops sending its request body is answered with a timeout', async (t) => {
+  let middleware;
+  const plugin = localMcpPlugin({
+    bodyTimeoutMs: 20,
+    createServer: () => ({ handle: async () => null }),
+  });
+  plugin.configureServer({
+    middlewares: { use: (path, handler) => (middleware = handler) },
+  });
+  const http = createServer((req, res) => middleware(req, res));
+  await new Promise((resolve) => http.listen(0, '127.0.0.1', resolve));
+  t.after(() => http.close());
+  const { port } = http.address();
+  const status = await new Promise((resolve, reject) => {
+    const request = httpRequest(
+      {
+        host: '127.0.0.1',
+        port,
+        path: '/mcp',
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+          'Content-Length': '100',
+          Host: `localhost:${port}`,
+        },
+      },
+      (response) => {
+        response.resume();
+        resolve(response.statusCode);
+        request.destroy();
+      },
+    );
+    request.on('error', (error) => {
+      if (error.code !== 'ECONNRESET') reject(error);
+    });
+    // Part of the body, then nothing more.
+    request.write('{"jsonrpc"');
+  });
+  assert.equal(status, 408);
+});

src/tools/localServer.test.mjs

@@ -171,3 +171,43 @@ test("only the key in the server's own panel page opens panel requests", async (
   });
   assert.doesNotMatch(otherPage.result.contents[0].text, new RegExp(key));
 });
+
+test('a request the client cancels over stdio is aborted and gets no answer', async () => {
+  let seen;
+  const server = {
+    handle: (message, { signal } = {}) => {
+      if (message.id === 1) {
+        seen = signal;
+        return new Promise((resolve) =>
+          signal.addEventListener(
+            'abort',
+            () => resolve({ jsonrpc: '2.0', id: 1, result: {} }),
+            { once: true },
+          ),
+        );
+      }
+      return Promise.resolve(
+        message.id === undefined
+          ? null
+          : { jsonrpc: '2.0', id: message.id, result: {} },
+      );
+    },
+  };
+  const input = new PassThrough();
+  const output = new PassThrough();
+  let written = '';
+  output.on('data', (chunk) => (written += chunk));
+  const served = serveStdio(server, { input, output });
+  input.write('{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{}}\n');
+  input.write('{"jsonrpc":"2.0","id":2,"method":"ping"}\n');
+  input.end(
+    '{"jsonrpc":"2.0","method":"notifications/cancelled","params":{"requestId":1}}\n',
+  );
+  await served;
+  assert.equal(seen.aborted, true);
+  const ids = written
+    .trim()
+    .split('\n')
+    .map((line) => JSON.parse(line).id);
+  assert.deepEqual(ids, [2]);
+});

src/tools/queries/panelRequest.js

@@ -48,9 +48,67 @@ const INCOMPRESSIBLE =
   /^(?:image\/(?!svg)|video\/|audio\/|font\/woff2)|zip|compressed/;
 const COMPRESS_MIN_BYTES = 1024;
 
-// Bodies too large for one call, kept for the calls that read the rest.
-const held = new Map();
-let heldBytes = 0;
+/**
+ * Requests in flight at once for one server's panels. Several panels (one
+ * per conversation) share a server, so further requests wait their turn
+ * rather than fail; past `PANEL_QUEUED_REQUESTS` waiting, they are refused.
+ */
+export const PANEL_CONCURRENT_REQUESTS = 6;
+export const PANEL_QUEUED_REQUESTS = 256;
+
+/**
+ * Each server's panel state: bodies too large for one call, kept for the
+ * calls that read the rest, and the requests in flight. Kept per server
+ * (its `services.app`), so one server's panel never reads another's.
+ */
+const panels = new WeakMap();
+function panelState(app) {
+  if (!panels.has(app))
+    panels.set(app, {
+      held: new Map(),
+      heldBytes: 0,
+      inFlight: 0,
+      waiting: [],
+    });
+  return panels.get(app);
+}
+
+/** Take a place among the requests in flight, waiting for one if need be. */
+async function acquire(state, signal) {
+  if (state.inFlight < PANEL_CONCURRENT_REQUESTS) {
+    state.inFlight += 1;
+    return;
+  }
+  if (state.waiting.length >= PANEL_QUEUED_REQUESTS)
+    throw new ToolError(
+      'retry_later',
+      'The panel has too many requests waiting',
+      {
+        retryAfterSeconds: 1,
+      },
+    );
+  await new Promise((resolve, reject) => {
+    const waiter = { resolve };
+    const abort = () => {
+      const index = state.waiting.indexOf(waiter);
+      if (index !== -1) state.waiting.splice(index, 1);
+      reject(signal.reason);
+    };
+    waiter.resolve = () => {
+      signal?.removeEventListener('abort', abort);
+      resolve();
+    };
+    signal?.addEventListener('abort', abort, { once: true });
+    state.waiting.push(waiter);
+  });
+}
+
+/** Give up a place, handing it straight to the next waiting request. */
+function releaseSlot(state) {
+  const next = state.waiting.shift();
+  if (next) next.resolve();
+  else state.inFlight -= 1;
+}
 
 function base64(bytes) {
   let binary = '';
@@ -81,26 +139,27 @@ async function gzip(bytes) {
   return new Uint8Array(await new Response(stream).arrayBuffer());
 }
 
-function release(id) {
-  heldBytes -= held.get(id).bytes.length;
-  held.delete(id);
+function release(state, id) {
+  state.heldBytes -= state.held.get(id).bytes.length;
+  state.held.delete(id);
 }
 
 /** Drop held responses nobody continued in time. */
-function forgetExpired(now = Date.now()) {
-  for (const [id, entry] of held) if (entry.expires <= now) release(id);
+function forgetExpired(state, now = Date.now()) {
+  for (const [id, entry] of state.held)
+    if (entry.expires <= now) release(state, id);
 }
 
 /** Hold a response for its later parts, first making room for it. */
-function hold(id, entry) {
-  forgetExpired();
+function hold(state, id, entry) {
+  forgetExpired(state);
   // Oldest first; a response being read is renewed, so it is the newest.
-  for (const oldest of held.keys()) {
-    if (heldBytes + entry.bytes.length <= HELD_LIMIT_BYTES) break;
-    release(oldest);
+  for (const oldest of state.held.keys()) {
+    if (state.heldBytes + entry.bytes.length <= HELD_LIMIT_BYTES) break;
+    release(state, oldest);
   }
-  held.set(id, entry);
-  heldBytes += entry.bytes.length;
+  state.held.set(id, entry);
+  state.heldBytes += entry.bytes.length;
 }
 
 /** Read a response body, refusing one larger than the panel may load. */
@@ -214,13 +273,14 @@ export const panelRequest = defineTool({
         'invalid_arguments',
         "Only the God's Eye View panel may make this request",
       );
-    forgetExpired();
+    const state = panelState(services.app);
+    forgetExpired(state);
     if (args.id !== undefined) {
-      const entry = held.get(args.id);
+      const entry = state.held.get(args.id);
       if (entry) {
         // Still being read: renew it, and keep it newest.
-        held.delete(args.id);
-        held.set(args.id, { ...entry, expires: Date.now() + HELD_MS });
+        state.held.delete(args.id);
+        state.held.set(args.id, { ...entry, expires: Date.now() + HELD_MS });
       }
       if (!entry)
         throw new ToolError(
@@ -241,6 +301,8 @@ export const panelRequest = defineTool({
       if (FORWARDED_REQUEST_HEADERS.includes(name.toLowerCase()))
         headers[name] = String(value);
     }
+    signal?.throwIfAborted();
+    await acquire(state, signal);
     const deadline = AbortSignal.timeout(REQUEST_TIMEOUT_MS);
     let answer;
     let bytes;
@@ -260,6 +322,8 @@ export const panelRequest = defineTool({
       if (signal?.aborted) throw error;
       if (error instanceof ToolError) throw error;
       throw new ToolError('unavailable', "The app's server did not answer");
+    } finally {
+      releaseSlot(state);
     }
     const type = answer.headers.get('content-type') || '';
     let encoding = 'identity';
@@ -280,7 +344,7 @@ export const panelRequest = defineTool({
     };
     if (bytes.length > PANEL_PART_BYTES) {
       response.id = crypto.randomUUID();
-      hold(response.id, {
+      hold(state, response.id, {
         path,
         response,
         bytes,

src/tools/queries/panelRequest.test.mjs

@@ -2,6 +2,7 @@ import assert from 'node:assert/strict';
 import test from 'node:test';
 import { composeCatalog, coreTools } from '../index.js';
 import {
+  PANEL_CONCURRENT_REQUESTS,
   PANEL_PART_BYTES,
   PANEL_RESPONSE_LIMIT_BYTES,
 } from './panelRequest.js';
@@ -251,3 +252,90 @@ test('routes the panel never loads are refused in any case or encoding', async (
     ['/api/setups-help', '/mcpx.json'],
   );
 });
+
+test('requests past the limit wait their turn instead of failing', async () => {
+  const gates = [];
+  const { catalog, requests } = catalogWith(
+    () =>
+      new Promise((resolve) => gates.push(() => resolve(new Response('ok')))),
+  );
+  const tick = () => new Promise((resolve) => setTimeout(resolve, 0));
+  // Two panels on one server, each sending as many requests as it may.
+  const calls = Array.from({ length: PANEL_CONCURRENT_REQUESTS * 2 }, (_, n) =>
+    catalog.call('panel_request', { key: KEY, path: `/panel/${n}.js` }),
+  );
+  await tick();
+  assert.equal(requests.length, PANEL_CONCURRENT_REQUESTS);
+  // Each finished request lets one waiting request start.
+  for (let started = PANEL_CONCURRENT_REQUESTS; gates.length;) {
+    gates.shift()();
+    await tick();
+    if (started < calls.length) {
+      started += 1;
+      assert.equal(requests.length, started);
+    }
+  }
+  const results = await Promise.all(calls);
+  assert.ok(results.every(({ data }) => data.status === 200));
+});
+
+test('a waiting request its caller cancels leaves the queue', async () => {
+  const gates = [];
+  const { catalog, requests } = catalogWith(
+    () =>
+      new Promise((resolve) => gates.push(() => resolve(new Response('ok')))),
+  );
+  const tick = () => new Promise((resolve) => setTimeout(resolve, 0));
+  const busy = Array.from({ length: PANEL_CONCURRENT_REQUESTS }, (_, n) =>
+    catalog.call('panel_request', { key: KEY, path: `/panel/${n}.js` }),
+  );
+  const cancel = new AbortController();
+  const waiting = catalog.call(
+    'panel_request',
+    { key: KEY, path: '/panel/cancelled.js' },
+    { signal: cancel.signal },
+  );
+  const after = catalog.call('panel_request', {
+    key: KEY,
+    path: '/panel/after.js',
+  });
+  await tick();
+  cancel.abort();
+  await assert.rejects(waiting);
+  gates.shift()();
+  await tick();
+  // The cancelled request never reached the server; the next one did.
+  assert.equal(requests.at(-1).path, '/panel/after.js');
+  while (gates.length) gates.shift()();
+  await Promise.all([...busy, after]);
+  assert.ok(
+    !requests.some((request) => request.path === '/panel/cancelled.js'),
+  );
+});
+
+test("one server's held responses are not readable through another's", async () => {
+  const big = new Uint8Array(PANEL_PART_BYTES * 2).fill(7);
+  const one = catalogWith(
+    () => new Response(big, { headers: { 'content-type': 'image/png' } }),
+  );
+  const other = catalogWith(() => new Response('x'));
+  const { data } = await one.catalog.call('panel_request', {
+    key: KEY,
+    path: '/big.png',
+  });
+  assert.ok(data.id);
+  await assert.rejects(
+    other.catalog.call('panel_request', {
+      key: KEY,
+      id: data.id,
+      offset: data.nextOffset,
+    }),
+    (error) => error.code === 'invalid_arguments',
+  );
+  const rest = await one.catalog.call('panel_request', {
+    key: KEY,
+    id: data.id,
+    offset: data.nextOffset,
+  });
+  assert.equal(rest.data.offset, data.nextOffset);
+});