fix: isolate stateless MCP requests

.changeset/fresh-mcp-transports.md

@@ -0,0 +1,5 @@
+---
+'incur': patch
+---
+
+Isolated stateless MCP requests, preventing cross-client response collisions and retained aborted requests.

src/Cli.test.ts

@@ -58,18 +58,20 @@ function mockMcpServeResponses(responses: unknown[]) {
   })
 }
 
-function countRetainedMcpResponses() {
-  return new Promise<number>((resolve, reject) => {
-    execFile(
-      process.execPath,
-      ['--expose-gc', '--import', 'tsx', 'test/fixtures/mcp-memory.ts'],
-      { cwd: join(import.meta.dirname, '..'), timeout: 30_000 },
-      (error, stdout, stderr) => {
-        if (error) reject(new Error(stderr.trim() || stdout.trim() || error.message))
-        else resolve(Number(stdout.trim()))
-      },
-    )
-  })
+function countRetainedMcpExchanges() {
+  return new Promise<{ aborted: number; requests: number; responses: number }>(
+    (resolve, reject) => {
+      execFile(
+        process.execPath,
+        ['--expose-gc', '--import', 'tsx', 'test/fixtures/mcp-memory.ts'],
+        { cwd: join(import.meta.dirname, '..'), timeout: 30_000 },
+        (error, stdout, stderr) => {
+          if (error) reject(new Error(stderr.trim() || stdout.trim() || error.message))
+          else resolve(JSON.parse(stdout.trim()))
+        },
+      )
+    },
+  )
 }
 
 function createConfigCli(flag?: string) {
@@ -5708,6 +5710,7 @@ describe('fetch', () => {
       body: unknown,
       sessionId?: string,
       extraHeaders: Record<string, string> = {},
+      signal?: AbortSignal,
     ) {
       const headers: Record<string, string> = {
         'content-type': 'application/json',
@@ -5720,6 +5723,7 @@ describe('fetch', () => {
           method: 'POST',
           headers,
           body: JSON.stringify(body),
+          ...(signal ? { signal } : {}),
         }),
       )
     }
@@ -5803,8 +5807,76 @@ describe('fetch', () => {
       `)
     })
 
-    test('completed JSON responses are released', async () => {
-      expect(await countRetainedMcpResponses()).toBe(0)
+    test('completed and aborted MCP exchanges are released', async () => {
+      expect(await countRetainedMcpExchanges()).toMatchInlineSnapshot(`
+        {
+          "aborted": 50,
+          "requests": 0,
+          "responses": 0,
+        }
+      `)
+    })
+
+    test('concurrent clients may reuse JSON-RPC ids', async () => {
+      let startFirst!: () => void
+      let startSecond!: () => void
+      let releaseFirst!: () => void
+      let releaseSecond!: () => void
+      const firstStarted = new Promise<void>((resolve) => (startFirst = resolve))
+      const secondStarted = new Promise<void>((resolve) => (startSecond = resolve))
+      const firstReleased = new Promise<void>((resolve) => (releaseFirst = resolve))
+      const secondReleased = new Promise<void>((resolve) => (releaseSecond = resolve))
+      const cli = Cli.create('test', {
+        version: '1.0.0',
+        mcp: { tools: { discovery: 'direct' } },
+      }).command('identify', {
+        run: async (c) => {
+          const authorization = c.request?.headers.get('authorization')
+          if (authorization === 'Bearer first') {
+            startFirst()
+            await firstReleased
+          } else {
+            startSecond()
+            await secondReleased
+          }
+          return { authorization }
+        },
+      })
+
+      const call = async (authorization: string) => {
+        const response = await mcpRequest(
+          cli,
+          {
+            jsonrpc: '2.0',
+            id: 1,
+            method: 'tools/call',
+            params: { name: 'identify', arguments: {} },
+          },
+          undefined,
+          { authorization },
+        )
+        const body = await response.json()
+        return JSON.parse(body.result.content[0].text)
+      }
+
+      const first = call('Bearer first')
+      await firstStarted
+      const second = call('Bearer second')
+      await secondStarted
+      releaseFirst()
+      await new Promise(setImmediate)
+      releaseSecond()
+
+      expect(await Promise.all([first, second])).toMatchInlineSnapshot(`
+        [
+          {
+            "authorization": "Bearer first",
+          },
+          {
+            "authorization": "Bearer second",
+          },
+        ]
+      `)
     })
 
     test('POST /mcp with tools/list → returns registered tools', async () => {
@@ -5916,6 +5988,51 @@ describe('fetch', () => {
       expect(await res.text()).toBe('')
     })
 
+    test('POST /mcp rejects pre-aborted requests', async () => {
+      const controller = new AbortController()
+      controller.abort()
+      await expect(
+        mcpRequest(
+          mcpCli(),
+          { jsonrpc: '2.0', id: 1, method: 'tools/list', params: {} },
+          undefined,
+          {},
+          controller.signal,
+        ),
+      ).rejects.toThrowErrorMatchingInlineSnapshot(`[AbortError: This operation was aborted]`)
+    })
+
+    test('POST /mcp rejects requests aborted during server startup', async () => {
+      let runs = 0
+      const cli = Cli.create('test', {
+        version: '1.0.0',
+        mcp: { tools: { discovery: 'direct' } },
+      }).command('run', {
+        run: () => {
+          runs++
+          return { ok: true }
+        },
+      })
+      const controller = new AbortController()
+      const response = mcpRequest(
+        cli,
+        {
+          jsonrpc: '2.0',
+          id: 1,
+          method: 'tools/call',
+          params: { name: 'run', arguments: {} },
+        },
+        undefined,
+        {},
+        controller.signal,
+      )
+      controller.abort()
+      await expect(response).rejects.toThrowErrorMatchingInlineSnapshot(
+        `[AbortError: This operation was aborted]`,
+      )
+      expect(runs).toMatchInlineSnapshot(`0`)
+    })
+
     test('mcp.stateless false keeps stateful session handling', async () => {
       const cli = Cli.create('test', { version: '1.0.0', mcp: { stateless: false } })
       cli.command('ping', {

src/Cli.ts

@@ -1824,7 +1824,46 @@ function createMcpHttpHandler(
   version: string,
   options: createMcpHttpHandler.Options = {},
 ) {
-  let transport: any
+  let session: ReturnType<typeof createServer> | undefined
+
+  async function createServer(
+    commands: Map<string, CommandEntry>,
+    mcpOptions:
+      | {
+          middlewares?: MiddlewareHandler[] | undefined
+          env?: z.ZodObject<any> | undefined
+          vars?: z.ZodObject<any> | undefined
+        }
+      | undefined,
+    stateless: boolean,
+  ) {
+    const { fromJsonSchema, McpServer, WebStandardStreamableHTTPServerTransport } =
+      await import('@modelcontextprotocol/server')
+
+    const server = new McpServer({ name, version })
+    Mcp.registerTools(server, commands, {
+      env: mcpOptions?.env,
+      fromJsonSchema,
+      middlewares: mcpOptions?.middlewares,
+      name,
+      request: (extra) => extra?.http?.req,
+      sendNotification: (notification) => server.server.notification(notification),
+      tools: options.tools,
+      vars: mcpOptions?.vars,
+      version,
+    })
+
+    const transport = new WebStandardStreamableHTTPServerTransport(
+      stateless
+        ? { enableJsonResponse: true }
+        : {
+            sessionIdGenerator: () => crypto.randomUUID(),
+            enableJsonResponse: true,
+          },
+    )
+    await server.connect(transport)
+    return { server, transport }
+  }
 
   return async (
     req: Request,
@@ -1839,33 +1878,43 @@ function createMcpHttpHandler(
     if (stateless && req.method !== 'POST')
       return new Response(null, { status: 405, headers: { Allow: 'POST' } })
 
-    if (!transport) {
-      const { fromJsonSchema, McpServer, WebStandardStreamableHTTPServerTransport } =
-        await import('@modelcontextprotocol/server')
-
-      const server = new McpServer({ name, version })
-      Mcp.registerTools(server, commands, {
-        env: mcpOptions?.env,
-        fromJsonSchema,
-        middlewares: mcpOptions?.middlewares,
-        name,
-        request: (extra) => extra?.http?.req,
-        sendNotification: (notification) => server.server.notification(notification),
-        tools: options.tools,
-        vars: mcpOptions?.vars,
-        version,
+    if (!stateless) {
+      session ??= createServer(commands, mcpOptions, false).catch((error) => {
+        session = undefined
+        throw error
       })
+      return (await session).transport.handleRequest(req)
+    }
 
-      const transportOptions = stateless
-        ? { enableJsonResponse: true }
-        : {
-            sessionIdGenerator: () => crypto.randomUUID(),
-            enableJsonResponse: true,
-          }
-      transport = new WebStandardStreamableHTTPServerTransport(transportOptions)
-      await server.connect(transport)
+    const abortReason = () =>
+      req.signal.reason ?? new DOMException('This operation was aborted', 'AbortError')
+    if (req.signal.aborted) throw abortReason()
+
+    const { server, transport } = await createServer(commands, mcpOptions, true)
+    let closing: Promise<void> | undefined
+    const close = () => (closing ??= server.close())
+    // Transport closure does not settle `handleRequest`; reject the public fetch separately.
+    let rejectAbort!: (reason?: unknown) => void
+    const aborted = new Promise<never>((_resolve, reject) => {
+      rejectAbort = reject
+    })
+    let didAbort = false
+    const abort = () => {
+      if (didAbort) return
+      didAbort = true
+      rejectAbort(abortReason())
+      void close()
+    }
+    req.signal.addEventListener('abort', abort, { once: true })
+    // Catch aborts that happened during asynchronous server creation.
+    if (req.signal.aborted) abort()
+    try {
+      if (req.signal.aborted) return await aborted
+      return await Promise.race([transport.handleRequest(req), aborted])
+    } finally {
+      req.signal.removeEventListener('abort', abort)
+      await close()
     }
-    return transport.handleRequest(req)
   }
 }
 

test/fixtures/mcp-memory.ts

@@ -1,13 +1,23 @@
 import { Cli } from '../../src/index.js'
 
+let started = 0
+const requests: WeakRef<Request>[] = []
 const cli = Cli.create('memory-test', {
   mcp: { tools: { discovery: 'direct' } },
   version: '1.0.0',
-}).command('ping', { run: () => ({ pong: true }) })
+})
+  .command('ping', { run: () => ({ pong: true }) })
+  .command('stall', {
+    run: (c) => {
+      if (c.request) requests.push(new WeakRef(c.request))
+      started++
+      return new Promise<never>(() => {})
+    },
+  })
 
 let id = 0
 
-function request(method: string, params: Record<string, unknown> = {}) {
+function request(method: string, params: Record<string, unknown> = {}, signal?: AbortSignal) {
   return cli.fetch(
     new Request('http://localhost/mcp', {
       body: JSON.stringify({ id: ++id, jsonrpc: '2.0', method, params }),
@@ -16,6 +26,7 @@ function request(method: string, params: Record<string, unknown> = {}) {
         'content-type': 'application/json',
       },
       method: 'POST',
+      ...(signal ? { signal } : {}),
     }),
   )
 }
@@ -40,11 +51,35 @@ await consume('initialize', {
 const responses: WeakRef<Response>[] = []
 for (let index = 0; index < 100; index++) responses.push(await weakResponse())
 
+let calls: Promise<boolean>[] | undefined = []
+let controllers: AbortController[] | undefined = []
+for (let index = 0; index < 50; index++) {
+  const controller = new AbortController()
+  controllers.push(controller)
+  calls.push(
+    request('tools/call', { name: 'stall', arguments: {} }, controller.signal).then(
+      () => false,
+      () => true,
+    ),
+  )
+}
+while (started < controllers.length) await new Promise(setImmediate)
+for (const controller of controllers) controller.abort()
+controllers = undefined
+const aborted = (await Promise.all(calls)).filter(Boolean).length
+calls = undefined
+
 const gc = globalThis.gc
 if (!gc) throw new Error('garbage collection is unavailable')
 for (let index = 0; index < 5; index++) {
   await new Promise(setImmediate)
   gc()
 }
 
-console.log(responses.filter((response) => response.deref()).length)
+console.log(
+  JSON.stringify({
+    aborted,
+    requests: requests.filter((request) => request.deref()).length,
+    responses: responses.filter((response) => response.deref()).length,
+  }),
+)