improvement(mcp): post-merge hardening — protocol negotiation + distributed OAuth lock + typed error dispatch

- Inbound MCP server now negotiates protocolVersion per MCP 2025-06-18: echoes
  the client's requested version when supported, falls back to our latest.
  Previously hardcoded to the oldest spec version (2024-11-05).
- withMcpOauthRefreshLock now takes a Postgres advisory transaction lock in
  addition to the in-process Promise chain, so concurrent processes (multi-task
  ECS) serialize on a per-OAuth-row basis. Previously a refresh race across
  processes could rotate a token under another process's feet and force re-auth.
- categorizeError dispatches on McpOauthAuthorizationRequiredError /
  UnauthorizedError / McpConnectionError first, only falling back to substring
  matching for SDK / third-party errors. Adds 502 for connection failures and
  503 for cooldown. Tests cover all four typed cases.
- discoverTools no longer pretends to handle deferred-side-effect rejections
  via a dead allSettled().catch() — each side-effect already self-logs; we just
  swallow per-promise to silence unhandled-rejection warnings.

Author: waleedlatif1

apps/sim/app/api/mcp/serve/[serverId]/route.test.ts

@@ -197,4 +197,49 @@ describe('MCP Serve Route', () => {
     expect(headers['X-API-Key']).toBeUndefined()
     expect(mockGenerateInternalToken).toHaveBeenCalledWith('user-1')
   })
+
+  describe('initialize protocol version negotiation', () => {
+    async function callInitialize(protocolVersion?: string) {
+      dbChainMockFns.limit.mockResolvedValueOnce([
+        {
+          id: 'server-1',
+          name: 'Public Server',
+          workspaceId: 'ws-1',
+          isPublic: true,
+          createdBy: 'owner-1',
+        },
+      ])
+      const params: Record<string, unknown> = {
+        capabilities: {},
+        clientInfo: { name: 'test', version: '1.0.0' },
+      }
+      if (protocolVersion !== undefined) params.protocolVersion = protocolVersion
+      const req = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
+        method: 'POST',
+        body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params }),
+      })
+      const res = await POST(req, { params: Promise.resolve({ serverId: 'server-1' }) })
+      return res.json() as Promise<{ result: { protocolVersion: string } }>
+    }
+
+    it('echoes a supported client protocolVersion (2025-06-18)', async () => {
+      const body = await callInitialize('2025-06-18')
+      expect(body.result.protocolVersion).toBe('2025-06-18')
+    })
+
+    it('echoes a supported client protocolVersion (2024-11-05)', async () => {
+      const body = await callInitialize('2024-11-05')
+      expect(body.result.protocolVersion).toBe('2024-11-05')
+    })
+
+    it('falls back to latest when client requests unknown version', async () => {
+      const body = await callInitialize('2099-01-01')
+      expect(body.result.protocolVersion).toBe('2025-06-18')
+    })
+
+    it('falls back to latest when client omits protocolVersion', async () => {
+      const body = await callInitialize(undefined)
+      expect(body.result.protocolVersion).toBe('2025-06-18')
+    })
+  })
 })

apps/sim/app/api/mcp/serve/[serverId]/route.ts

@@ -36,6 +36,22 @@ import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('WorkflowMcpServeAPI')
 
+// Newest first. We echo the client's version when we support it (per
+// MCP 2025-06-18 lifecycle spec); otherwise fall back to our latest.
+const SUPPORTED_PROTOCOL_VERSIONS = ['2025-06-18', '2025-03-26', '2024-11-05'] as const
+const LATEST_PROTOCOL_VERSION = SUPPORTED_PROTOCOL_VERSIONS[0]
+
+function negotiateProtocolVersion(rpcParams: unknown): string {
+  const requested =
+    rpcParams && typeof rpcParams === 'object' && 'protocolVersion' in rpcParams
+      ? (rpcParams as { protocolVersion?: unknown }).protocolVersion
+      : undefined
+  if (typeof requested === 'string' && SUPPORTED_PROTOCOL_VERSIONS.includes(requested as never)) {
+    return requested
+  }
+  return LATEST_PROTOCOL_VERSION
+}
+
 export const dynamic = 'force-dynamic'
 
 interface RouteParams {
@@ -214,7 +230,7 @@ export const POST = withRouteHandler(
       switch (method) {
         case 'initialize': {
           const result: InitializeResult = {
-            protocolVersion: '2024-11-05',
+            protocolVersion: negotiateProtocolVersion(rpcParams),
             capabilities: { tools: {} },
             serverInfo: { name: server.name, version: '1.0.0' },
           }

apps/sim/lib/mcp/oauth/storage.ts

@@ -8,7 +8,7 @@ import { mcpServerOauth } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
-import { and, eq, gt } from 'drizzle-orm'
+import { and, eq, gt, sql } from 'drizzle-orm'
 import { decryptSecret, encryptSecret } from '@/lib/core/security/encryption'
 
 const logger = createLogger('McpOauthStorage')
@@ -227,19 +227,31 @@ export async function clearState(rowId: string): Promise<void> {
 }
 
 /**
- * Per-process serialization for an OAuth row. Refresh tokens rotate (RFC 6749 §6,
- * MCP §2.3.3), so two concurrent refreshes against the same row would race and one
- * would receive `invalid_grant`, wiping the credentials. We serialize SDK calls
- * that may trigger a refresh on a per-row basis.
+ * Serialize OAuth row access across all callers, in-process AND across
+ * processes. Refresh tokens rotate (RFC 6749 §6, MCP §2.3.3), so two concurrent
+ * refreshes against the same row would race and one would receive
+ * `invalid_grant`, wiping credentials.
+ *
+ * Two-tier locking:
+ *   1) In-process Promise chain — cheap, avoids DB roundtrips when the same
+ *      Node process holds concurrent callers.
+ *   2) Postgres advisory transaction lock — blocks across processes; released
+ *      automatically when the transaction ends.
  */
 const refreshLocks = new Map<string, Promise<unknown>>()
 
 export async function withMcpOauthRefreshLock<T>(rowId: string, fn: () => Promise<T>): Promise<T> {
   const prev = refreshLocks.get(rowId) ?? Promise.resolve()
-  // Wait for the predecessor to settle (success or failure), discard its
-  // value/error, then run fn. Each caller awaits its own fn's outcome — errors
-  // do not propagate across callers in the chain.
-  const next = prev.catch(() => undefined).then(() => fn())
+  const next = prev
+    .catch(() => undefined)
+    .then(() =>
+      db.transaction(async (tx) => {
+        await tx.execute(
+          sql`SELECT pg_advisory_xact_lock(hashtextextended(${`mcp_oauth_refresh:${rowId}`}, 0))`
+        )
+        return fn()
+      })
+    )
   refreshLocks.set(rowId, next)
   const cleanup = () => {
     if (refreshLocks.get(rowId) === next) refreshLocks.delete(rowId)

apps/sim/lib/mcp/service.ts

@@ -601,9 +601,9 @@ class McpService {
 
       // Await cache writes so a follow-up discoverTools sees consistent state.
       await Promise.allSettled(cacheWrites)
-      Promise.allSettled(deferredSideEffects).catch((err) => {
-        logger.error(`[${requestId}] Error in deferred discovery work:`, err)
-      })
+      // Each deferred side-effect self-logs failures, so we just mark the
+      // promises as handled to avoid unhandled-rejection warnings.
+      for (const p of deferredSideEffects) p.catch(() => {})
 
       if (mcpConnectionManager) {
         for (const conn of liveConnections) {

apps/sim/lib/mcp/utils.test.ts

@@ -1,5 +1,7 @@
+import { UnauthorizedError } from '@modelcontextprotocol/sdk/client/auth.js'
 import { describe, expect, it } from 'vitest'
 import { DEFAULT_EXECUTION_TIMEOUT_MS } from '@/lib/core/execution-limits'
+import { McpConnectionError, McpOauthAuthorizationRequiredError } from '@/lib/mcp/types'
 import {
   categorizeError,
   createMcpToolId,
@@ -304,6 +306,32 @@ describe('categorizeError', () => {
     expect(result.message).toBe('Unknown error occurred')
   })
 
+  it.concurrent('returns 401 for McpOauthAuthorizationRequiredError via instanceof', () => {
+    const error = new McpOauthAuthorizationRequiredError('mcp-a', 'A')
+    const result = categorizeError(error)
+    expect(result.status).toBe(401)
+    expect(result.message).toBe('Authentication required')
+  })
+
+  it.concurrent('returns 401 for SDK UnauthorizedError via instanceof', () => {
+    const error = new UnauthorizedError('token expired')
+    const result = categorizeError(error)
+    expect(result.status).toBe(401)
+  })
+
+  it.concurrent('returns 503 for McpConnectionError with cooldown message', () => {
+    const error = new McpConnectionError('Server in cooldown — try again shortly.', 'mcp-a')
+    const result = categorizeError(error)
+    expect(result.status).toBe(503)
+  })
+
+  it.concurrent('returns 502 for other McpConnectionError', () => {
+    const error = new McpConnectionError('connect ECONNREFUSED', 'mcp-a')
+    const result = categorizeError(error)
+    expect(result.status).toBe(502)
+    expect(result.message).toBe('Connection failed')
+  })
+
   it.concurrent('returns 500 for null', () => {
     const result = categorizeError(null)
     expect(result.status).toBe(500)

apps/sim/lib/mcp/utils.ts

@@ -1,6 +1,11 @@
+import { UnauthorizedError } from '@modelcontextprotocol/sdk/client/auth.js'
 import { NextResponse } from 'next/server'
 import { DEFAULT_EXECUTION_TIMEOUT_MS } from '@/lib/core/execution-limits'
-import type { McpApiResponse } from '@/lib/mcp/types'
+import {
+  type McpApiResponse,
+  McpConnectionError,
+  McpOauthAuthorizationRequiredError,
+} from '@/lib/mcp/types'
 import { isMcpTool, MCP } from '@/executor/constants'
 
 export const MCP_CONSTANTS = {
@@ -137,28 +142,36 @@ export function categorizeError(error: unknown): { message: string; status: numb
     return { message: 'Unknown error occurred', status: 500 }
   }
 
+  // Typed dispatch first — our own classes carry definitive intent.
+  if (error instanceof McpOauthAuthorizationRequiredError || error instanceof UnauthorizedError) {
+    return { message: 'Authentication required', status: 401 }
+  }
+  if (error instanceof McpConnectionError) {
+    if (error.message.toLowerCase().includes('cooldown')) {
+      return { message: 'Server temporarily unavailable', status: 503 }
+    }
+    return { message: 'Connection failed', status: 502 }
+  }
+
+  // Fall back to substring matching for SDK / third-party errors we don't
+  // own a typed class for.
   const msg = error.message.toLowerCase()
 
   if (msg.includes('timeout')) {
     return { message: 'Request timed out', status: 408 }
   }
-
   if (msg.includes('cooldown')) {
     return { message: 'Server temporarily unavailable', status: 503 }
   }
-
   if (msg.includes('not found') || msg.includes('not accessible')) {
     return { message: 'Resource not found', status: 404 }
   }
-
   if (msg.includes('authentication') || msg.includes('unauthorized')) {
     return { message: 'Authentication required', status: 401 }
   }
-
   if (msg.includes('invalid') || msg.includes('missing required') || msg.includes('validation')) {
     return { message: 'Invalid request parameters', status: 400 }
   }
-
   return { message: 'Internal server error', status: 500 }
 }