fix(mcp): Avoid leaked idle request tracking

Avoid inflating the MCP idle in-flight counter when a client reuses a pending
JSON-RPC request id. Also clear the pending request if downstream message
handling throws synchronously before a response can be sent.

Refs #394
Co-Authored-By: Codex <noreply@openai.com>

Author: cameroncooke

src/server/__tests__/server.test.ts

@@ -101,14 +101,20 @@ describe('MCP server transport request lifecycle instrumentation', () => {
     expect(onRequestCompleted).not.toHaveBeenCalled();
   });
 
-  it('completes duplicate responses only once', async () => {
+  it('ignores duplicate request IDs until the pending request completes', async () => {
+    const onRequestStarted = vi.fn();
     const onRequestCompleted = vi.fn();
-    const { transport } = await createStartedInstrumentedTransport({ onRequestCompleted });
+    const { transport } = await createStartedInstrumentedTransport({
+      onRequestStarted,
+      onRequestCompleted,
+    });
 
     transport.onmessage?.({ jsonrpc: '2.0', id: '1', method: 'initialize' });
+    transport.onmessage?.({ jsonrpc: '2.0', id: '1', method: 'tools/list' });
     await transport.send({ jsonrpc: '2.0', id: '1', result: {} });
     await transport.send({ jsonrpc: '2.0', id: '1', result: {} });
 
+    expect(onRequestStarted).toHaveBeenCalledTimes(1);
     expect(onRequestCompleted).toHaveBeenCalledTimes(1);
   });
 
@@ -124,4 +130,26 @@ describe('MCP server transport request lifecycle instrumentation', () => {
 
     expect(onRequestCompleted).toHaveBeenCalledTimes(1);
   });
+
+  it('marks completion when downstream message handling throws synchronously', async () => {
+    const onRequestStarted = vi.fn();
+    const onRequestCompleted = vi.fn();
+    const transport = new TestTransport();
+    const downstreamOnMessage = vi.fn(() => {
+      throw new Error('handler failed');
+    });
+    instrumentMcpRequestLifecycle(transport, { onRequestStarted, onRequestCompleted });
+
+    transport.onmessage = downstreamOnMessage;
+    await transport.start();
+
+    expect(() => {
+      transport.onmessage?.({ jsonrpc: '2.0', id: '1', method: 'tools/list' });
+    }).toThrow('handler failed');
+
+    expect(onRequestStarted).toHaveBeenCalledTimes(1);
+    expect(onRequestCompleted).toHaveBeenCalledTimes(1);
+    await transport.send({ jsonrpc: '2.0', id: '1', result: {} });
+    expect(onRequestCompleted).toHaveBeenCalledTimes(1);
+  });
 });

src/server/request-lifecycle.ts

@@ -47,12 +47,25 @@ export function instrumentMcpRequestLifecycle(
     onMessageWrapped = true;
     const downstreamOnMessage = transport.onmessage;
     transport.onmessage = (message, extra) => {
+      let startedRequestId: string | null = null;
+
       if (isJSONRPCRequest(message)) {
-        pendingRequestIds.add(requestIdKey(message.id));
-        observer.onRequestStarted?.();
+        const requestId = requestIdKey(message.id);
+        if (!pendingRequestIds.has(requestId)) {
+          pendingRequestIds.add(requestId);
+          startedRequestId = requestId;
+          observer.onRequestStarted?.();
+        }
       }
 
-      downstreamOnMessage(message, extra);
+      try {
+        downstreamOnMessage(message, extra);
+      } catch (error) {
+        if (startedRequestId !== null && pendingRequestIds.delete(startedRequestId)) {
+          observer.onRequestCompleted?.();
+        }
+        throw error;
+      }
     };
   };