fix(mcp): bugbot — TTL watchdog on OAuth lock + don't close evicted pinned agents

- Redis refresh lock now uses a 15s TTL with a watchdog that extends every
  5s while fn() runs. Long-running OAuth refreshes no longer lose the lock
  mid-flight and let another process race the same refresh.
- Pinned-agent LRU eviction no longer calls `agent.close()`. Existing
  `createMcpPinnedFetch` closures hold the dispatcher reference and were
  using a closed Agent after eviction. We drop from the cache and let GC
  release the dispatcher once the last closure dies; undici closes idle
  keep-alive connections via its own internal timeout.
- New tests: watchdog extends while fn() runs and stops once it settles;
  evicted agents are not closed and captured closures still work.

Author: waleedlatif1

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

@@ -11,9 +11,10 @@ import {
 } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockAcquireLock, mockReleaseLock } = vi.hoisted(() => ({
+const { mockAcquireLock, mockReleaseLock, mockExtendLock } = vi.hoisted(() => ({
   mockAcquireLock: vi.fn(),
   mockReleaseLock: vi.fn(),
+  mockExtendLock: vi.fn(),
 }))
 
 vi.mock('@sim/db', () => dbChainMock)
@@ -22,6 +23,7 @@ vi.mock('@/lib/core/security/encryption', () => encryptionMock)
 vi.mock('@/lib/core/config/redis', () => ({
   acquireLock: mockAcquireLock,
   releaseLock: mockReleaseLock,
+  extendLock: mockExtendLock,
 }))
 
 import {
@@ -112,7 +114,9 @@ describe('withMcpOauthRefreshLock', () => {
     vi.clearAllMocks()
     mockAcquireLock.mockReset()
     mockReleaseLock.mockReset()
+    mockExtendLock.mockReset()
     mockReleaseLock.mockResolvedValue(true)
+    mockExtendLock.mockResolvedValue(true)
   })
 
   it('serializes concurrent in-process callers, each running its own fn()', async () => {
@@ -197,4 +201,37 @@ describe('withMcpOauthRefreshLock', () => {
     expect(keys).toContain('mcp:oauth:refresh:row-a')
     expect(keys).toContain('mcp:oauth:refresh:row-b')
   })
+
+  it('extends the lock TTL while fn() is running so long refreshes do not lose the lock', async () => {
+    vi.useFakeTimers()
+    try {
+      mockAcquireLock.mockResolvedValue(true)
+      let resolveFn: (v: string) => void
+      const fn = vi.fn(
+        () =>
+          new Promise<string>((resolve) => {
+            resolveFn = resolve
+          })
+      )
+
+      const pending = withMcpOauthRefreshLock('row-watchdog', fn)
+
+      // Advance time past two extend intervals (5s + 5s = 10s).
+      await vi.advanceTimersByTimeAsync(11_000)
+      expect(mockExtendLock.mock.calls.length).toBeGreaterThanOrEqual(2)
+      for (const call of mockExtendLock.mock.calls) {
+        expect(call[0]).toBe('mcp:oauth:refresh:row-watchdog')
+      }
+
+      resolveFn!('done')
+      await expect(pending).resolves.toBe('done')
+
+      // Watchdog must stop once fn() settles — no more extend calls.
+      const extendCallsAtFinish = mockExtendLock.mock.calls.length
+      await vi.advanceTimersByTimeAsync(20_000)
+      expect(mockExtendLock.mock.calls.length).toBe(extendCallsAtFinish)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
 })

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

@@ -10,7 +10,7 @@ import { toError } from '@sim/utils/errors'
 import { sleep } from '@sim/utils/helpers'
 import { generateId, generateShortId } from '@sim/utils/id'
 import { and, eq, gt } from 'drizzle-orm'
-import { acquireLock, releaseLock } from '@/lib/core/config/redis'
+import { acquireLock, extendLock, releaseLock } from '@/lib/core/config/redis'
 import { decryptSecret, encryptSecret } from '@/lib/core/security/encryption'
 
 const logger = createLogger('McpOauthStorage')
@@ -238,13 +238,16 @@ export async function clearState(rowId: string): Promise<void> {
  * `McpClient` instances that can't be shared, unlike a scalar access token):
  *   1) In-process: per-row Promise chain. Concurrent callers queue; each
  *      runs `fn()` after the previous settles.
- *   2) Cross-process: Redis mutex (`acquireLock` / `releaseLock`). Followers
- *      poll until the holder releases, then acquire and run `fn()`.
+ *   2) Cross-process: Redis mutex (`acquireLock` / `releaseLock`) with a TTL
+ *      watchdog that periodically extends the lock while `fn()` runs, so
+ *      long-running refreshes don't drop the lock and let another process
+ *      race onto the same refresh.
  *
  * Falls open if Redis is unavailable — `acquireLock` no-ops, but in-process
  * serialization still holds within a single Node process.
  */
-const REFRESH_LOCK_TTL_SEC = 30
+const REFRESH_LOCK_TTL_SEC = 15
+const REFRESH_LOCK_EXTEND_INTERVAL_MS = 5_000
 const REFRESH_POLL_INTERVAL_MS = 100
 const REFRESH_MAX_WAIT_MS = 30_000
 
@@ -283,9 +286,18 @@ async function runWithRedisMutex<T>(
     }
 
     if (acquired) {
+      const watchdog = setInterval(() => {
+        extendLock(lockKey, ownerToken, REFRESH_LOCK_TTL_SEC).catch((error) => {
+          logger.warn('Refresh lock extend failed', {
+            rowId,
+            error: toError(error).message,
+          })
+        })
+      }, REFRESH_LOCK_EXTEND_INTERVAL_MS)
       try {
         return await fn()
       } finally {
+        clearInterval(watchdog)
         await releaseLock(lockKey, ownerToken).catch((error) => {
           logger.warn('Refresh lock release failed (will expire via TTL)', {
             rowId,

apps/sim/lib/mcp/pinned-fetch.test.ts

@@ -3,22 +3,27 @@
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockAgent, mockCreatePinnedLookup, mockUndiciFetch, capturedAgentOptions } = vi.hoisted(
-  () => {
+const { mockAgent, mockCreatePinnedLookup, mockUndiciFetch, capturedAgentOptions, agentCloses } =
+  vi.hoisted(() => {
     const capturedAgentOptions: unknown[] = []
+    const agentCloses: unknown[] = []
     class MockAgent {
       constructor(options: unknown) {
         capturedAgentOptions.push(options)
       }
+      close() {
+        agentCloses.push(this)
+        return Promise.resolve()
+      }
     }
     return {
       mockAgent: MockAgent,
       mockCreatePinnedLookup: vi.fn(),
       mockUndiciFetch: vi.fn(),
       capturedAgentOptions,
+      agentCloses,
     }
-  }
-)
+  })
 
 vi.mock('undici', () => ({ Agent: mockAgent, fetch: mockUndiciFetch }))
 vi.mock('@/lib/core/security/input-validation.server', () => ({
@@ -31,6 +36,7 @@ describe('createMcpPinnedFetch', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     capturedAgentOptions.length = 0
+    agentCloses.length = 0
     __resetPinnedAgentsForTests()
     mockCreatePinnedLookup.mockReturnValue('pinned-lookup-fn')
     mockUndiciFetch.mockResolvedValue(new Response('ok'))
@@ -105,4 +111,17 @@ describe('createMcpPinnedFetch', () => {
     const d2 = (mockUndiciFetch.mock.calls[1][1] as { dispatcher: unknown }).dispatcher
     expect(d1).not.toBe(d2)
   })
+
+  it('does not close evicted agents — captured closures keep working', async () => {
+    // Build an early closure whose agent will get evicted by later IPs.
+    const earlyClient = createMcpPinnedFetch('10.0.0.1')
+    // Fill the cache past its 64-entry limit so the early entry is evicted.
+    for (let i = 0; i < 64; i++) createMcpPinnedFetch(`10.1.${Math.floor(i / 256)}.${i % 256}`)
+
+    // Eviction must NOT have closed any agents.
+    expect(agentCloses).toHaveLength(0)
+    // The early closure's captured dispatcher is still callable.
+    await earlyClient('https://example.com/still-works')
+    expect(mockUndiciFetch).toHaveBeenCalledTimes(1)
+  })
 })

apps/sim/lib/mcp/pinned-fetch.ts

@@ -23,24 +23,20 @@ function getPinnedAgent(resolvedIP: string): Agent {
     return existing
   }
   if (pinnedAgents.size >= MAX_POOLED_AGENTS) {
+    // Drop the oldest entry WITHOUT closing it — existing `createMcpPinnedFetch`
+    // closures may still hold a reference and have in-flight requests. The
+    // dispatcher is GC'd (and its sockets cleaned up) when the last closure
+    // releases it; undici closes idle keep-alive connections after its own
+    // timeout (default 4s).
     const oldestKey = pinnedAgents.keys().next().value
-    if (oldestKey !== undefined) {
-      const oldest = pinnedAgents.get(oldestKey)
-      pinnedAgents.delete(oldestKey)
-      oldest?.close().catch(() => {})
-    }
+    if (oldestKey !== undefined) pinnedAgents.delete(oldestKey)
   }
   const agent = new Agent({ connect: { lookup: createPinnedLookup(resolvedIP) } })
   pinnedAgents.set(resolvedIP, agent)
   return agent
 }
 
 export function __resetPinnedAgentsForTests(): void {
-  for (const agent of pinnedAgents.values()) {
-    try {
-      void agent.close?.()
-    } catch {}
-  }
   pinnedAgents.clear()
 }