fix(mcp): use threading.Lock for OAuth token refresh to avoid cross-thread deadlock (#4240)

* fix(mcp): use threading.Lock for OAuth token refresh to avoid cross-thread deadlock

OAuthTokenManager created one asyncio.Lock per server for the process
lifetime. The embedded/TUI sync tool-call path (DeerFlowClient.stream()
-> LangGraph's ToolNode._func -> a ThreadPoolExecutor ->
make_sync_tool_wrapper's per-call asyncio.run()) invokes
get_authorization_header from a fresh event loop on a fresh OS thread
for every concurrent tool call. asyncio.Lock binds to whichever loop
first contends on it; when a caller on a different loop later releases
or wakes a waiter, it does so without call_soon_threadsafe, so the
waiting loop's selector is never woken and that caller hangs forever
with no exception. A third concurrent caller instead raises a
synchronous RuntimeError ("bound to a different event loop"). Either
way, two concurrent OAuth-protected tool calls (including the very
first cold-start token fetch) can freeze the entire agent turn.
Gateway's async path (ToolNode._afunc) is unaffected.

Replace the asyncio.Lock with a plain threading.Lock, acquired via
asyncio.to_thread so the blocking wait never blocks the event loop,
and released synchronously in a finally block. This keeps the
single-fetch de-duplication the lock provided while making it safe
across however many event loops/threads call into the same server's
lock.

Adds a regression test that runs three threads, each with its own
event loop, calling get_authorization_header concurrently for the same
server, and asserts (with a bounded join timeout so a regression fails
fast instead of hanging the suite) that none hang or raise, and that
only one real token fetch happens.

* fix(mcp): make OAuth lock acquisition cancellation-safe

get_authorization_header acquired the per-server threading.Lock via a
bare `await asyncio.to_thread(lock.acquire)`, with the try/finally that
guarantees release only starting after that await returned. Once the
executor thread had actually started running lock.acquire(), cancelling
the awaiting caller only stopped the caller -- Python cannot interrupt a
running OS thread. CancelledError was still delivered to the caller
immediately, but the thread kept blocking until the current holder
released, then silently acquired the lock with nobody left to call
release() for it. The lock stayed locked forever and every later OAuth
token refresh for that server blocked permanently at the same line --
the exact cross-thread deadlock this lock was introduced to prevent,
reintroduced via a different path under cancellation (e.g. a caller
wrapped in asyncio.wait_for/asyncio.timeout, or task-group cancellation).

Run the acquisition as an explicit asyncio.create_task, awaited via
asyncio.shield() so cancelling the caller no longer cancels the
underlying acquisition task. If the caller is cancelled, keep
(re-)waiting on the still-shielded acquisition task -- tolerating
further cancellation during this cleanup by simply retrying -- until it
actually finishes, release the lock immediately, and only then
re-raise. This guarantees the lock is released regardless of when or
how many times the caller is cancelled: before the acquisition is even
scheduled, while queued, or after it has already been silently granted.

Adds a regression test that holds the per-server lock, starts a second
caller that has to wait for it, cancels that caller while it is
genuinely blocked in its executor thread, releases the original holder,
and asserts a third caller completes within a bounded asyncio.wait_for
and still performs exactly one token fetch. Every potentially-hanging
await is bounded so a regression fails the test quickly instead of
hanging the suite.

backend/packages/harness/deerflow/mcp/oauth.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import threading
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from typing import Any
@@ -28,7 +29,17 @@ class OAuthTokenManager:
     def __init__(self, oauth_by_server: dict[str, McpOAuthConfig]):
         self._oauth_by_server = oauth_by_server
         self._tokens: dict[str, _OAuthToken] = {}
-        self._locks: dict[str, asyncio.Lock] = {name: asyncio.Lock() for name in oauth_by_server}
+        # A plain threading.Lock, not asyncio.Lock: the embedded/TUI sync tool-call
+        # path (DeerFlowClient.stream() -> LangGraph ToolNode._func -> a
+        # ThreadPoolExecutor -> deerflow.tools.sync.make_sync_tool_wrapper's
+        # per-call asyncio.run()) invokes get_authorization_header from a fresh
+        # event loop on a fresh OS thread for every concurrent tool call. An
+        # asyncio.Lock binds to whichever loop first contends on it; a second
+        # caller's release/wake-up crossing loops without call_soon_threadsafe
+        # either deadlocks silently or raises "bound to a different event loop".
+        # threading.Lock has no loop affinity, so it is safe to share across
+        # however many event loops/threads call into the same server's lock.
+        self._locks: dict[str, threading.Lock] = {name: threading.Lock() for name in oauth_by_server}
 
     @classmethod
     def from_extensions_config(cls, extensions_config: ExtensionsConfig) -> OAuthTokenManager:
@@ -54,7 +65,40 @@ class OAuthTokenManager:
             return f"{token.token_type} {token.access_token}"
 
         lock = self._locks[server_name]
-        async with lock:
+        # Acquire the OS-level lock off-thread so a blocking wait never blocks this
+        # event loop, then release it synchronously (release() never blocks). This
+        # keeps the de-duplication behavior of the old `async with lock:` (only one
+        # concurrent caller per server actually fetches a token) while remaining
+        # safe when callers are on different event loops/threads.
+        #
+        # The acquisition itself runs as an explicit Task, shielded from this
+        # coroutine's own cancellation. A bare `await asyncio.to_thread(lock.acquire)`
+        # cannot be safely cancelled: once the executor thread has started running
+        # lock.acquire(), Python has no way to stop it, so a cancellation delivered
+        # at that await would still let the thread go on to acquire the lock later
+        # (whenever the current holder releases it) with this coroutine already
+        # gone and nobody left to call release() -- the lock would stay locked
+        # forever and every later call for this server would block permanently at
+        # this same line. Shielding the acquisition task means a cancelled caller
+        # can instead wait for that (unstoppable) acquisition to actually land and
+        # release the lock immediately, rather than leaking ownership of it.
+        acquire_task = asyncio.create_task(asyncio.to_thread(lock.acquire), name=f"oauth-lock-acquire:{server_name}")
+        try:
+            await asyncio.shield(acquire_task)
+        except asyncio.CancelledError:
+            # Keep waiting -- shielded on every retry -- until the acquisition
+            # actually finishes, even if this coroutine is cancelled again while
+            # cleaning up: the underlying thread cannot be interrupted, so this is
+            # the only way to learn when the lock becomes ours and release it
+            # right away instead of leaving it locked forever.
+            while not acquire_task.done():
+                try:
+                    await asyncio.shield(acquire_task)
+                except asyncio.CancelledError:
+                    continue
+            lock.release()
+            raise
+        try:
             token = self._tokens.get(server_name)
             if token and not self._is_expiring(token, oauth):
                 return f"{token.token_type} {token.access_token}"
@@ -63,6 +107,8 @@ class OAuthTokenManager:
             self._tokens[server_name] = fresh
             logger.info(f"Refreshed OAuth access token for MCP server: {server_name}")
             return f"{fresh.token_type} {fresh.access_token}"
+        finally:
+            lock.release()
 
     @staticmethod
     def _is_expiring(token: _OAuthToken, oauth: McpOAuthConfig) -> bool:

backend/tests/test_mcp_oauth.py

@@ -3,8 +3,11 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 from typing import Any
 
+import pytest
+
 from deerflow.config.extensions_config import ExtensionsConfig
 from deerflow.mcp.oauth import OAuthTokenManager, build_oauth_tool_interceptor, get_initial_oauth_headers
 
@@ -332,3 +335,229 @@ def test_oauth_refresh_token_rotation_persists_rotated_value(monkeypatch):
     assert second == "Bearer at-1"
     assert len(post_calls) == 2
     assert post_calls[1]["data"]["refresh_token"] == "rt-rotated-1"
+
+
+def test_get_authorization_header_concurrent_threads_no_deadlock(monkeypatch):
+    """Concurrent callers on different event loops/threads must not deadlock.
+
+    The embedded/TUI sync tool-call path (``DeerFlowClient.stream()`` ->
+    LangGraph's ``ToolNode._func`` -> a ``ThreadPoolExecutor`` ->
+    ``deerflow.tools.sync.make_sync_tool_wrapper``'s per-call ``asyncio.run()``)
+    invokes ``get_authorization_header`` from a fresh event loop on a fresh OS
+    thread for every concurrent tool call. A per-server ``asyncio.Lock`` binds
+    to whichever loop first contends on it; when a caller on a *different*
+    loop later releases/wakes a waiter, it does so without
+    ``call_soon_threadsafe``, so the waiting loop's selector is never woken
+    and that caller hangs forever with no exception (a silent hang). A third
+    concurrent caller instead hits a synchronous ``RuntimeError: ... is bound
+    to a different event loop``. Both failure modes are reproducible with the
+    old ``asyncio.Lock``-per-server implementation.
+
+    This test uses a bounded thread-join timeout so that a regression back to
+    the old behavior fails this test quickly instead of hanging the whole
+    suite.
+    """
+    post_calls: list[dict[str, Any]] = []
+    post_calls_guard = threading.Lock()
+    holder_in_critical_section = threading.Event()
+
+    class _SlowMockAsyncClient:
+        def __init__(self, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        async def post(self, url: str, data: dict[str, Any]):
+            with post_calls_guard:
+                post_calls.append({"url": url, "data": data})
+            # Signal that this call is inside the critical section (the lock
+            # is held) and stay there briefly so the other threads have time
+            # to reach their own acquire() and genuinely contend, rather than
+            # racing to also take an uncontended fast path.
+            holder_in_critical_section.set()
+            await asyncio.sleep(0.3)
+            return _MockResponse(
+                {
+                    "access_token": "concurrent-token",
+                    "token_type": "Bearer",
+                    "expires_in": 3600,
+                }
+            )
+
+    monkeypatch.setattr("httpx.AsyncClient", _SlowMockAsyncClient)
+
+    config = ExtensionsConfig.model_validate(
+        {
+            "mcpServers": {
+                "secure-http": {
+                    "enabled": True,
+                    "type": "http",
+                    "url": "https://api.example.com/mcp",
+                    "oauth": {
+                        "enabled": True,
+                        "token_url": "https://auth.example.com/oauth/token",
+                        "grant_type": "client_credentials",
+                        "client_id": "client-id",
+                        "client_secret": "client-secret",
+                    },
+                }
+            }
+        }
+    )
+
+    manager = OAuthTokenManager.from_extensions_config(config)
+    results: dict[str, Any] = {}
+
+    def run_in_own_loop(name: str, wait_for_holder: bool) -> None:
+        if wait_for_holder:
+            # Only start once another thread is confirmed to be holding the
+            # lock, guaranteeing this call contends instead of racing for
+            # the uncontended fast path itself.
+            assert holder_in_critical_section.wait(timeout=5), "holder thread never entered critical section"
+        try:
+            results[name] = asyncio.run(manager.get_authorization_header("secure-http"))
+        except BaseException as exc:  # noqa: BLE001 - captured to assert absence below
+            results[name] = exc
+
+    threads = [
+        threading.Thread(target=run_in_own_loop, args=("holder", False), name="holder", daemon=True),
+        threading.Thread(target=run_in_own_loop, args=("waiter-1", True), name="waiter-1", daemon=True),
+        threading.Thread(target=run_in_own_loop, args=("waiter-2", True), name="waiter-2", daemon=True),
+    ]
+
+    for t in threads:
+        t.start()
+
+    # Bounded timeout: under the old per-server asyncio.Lock, at least one of
+    # these threads would never return. Joining with a timeout keeps a
+    # regression from hanging the test suite forever; it fails fast instead.
+    for t in threads:
+        t.join(timeout=5)
+
+    still_alive = [t.name for t in threads if t.is_alive()]
+    assert not still_alive, f"deadlock: thread(s) still blocked after bounded timeout: {still_alive}"
+
+    for name, result in results.items():
+        assert not isinstance(result, BaseException), f"{name} raised instead of completing: {result!r}"
+        assert result == "Bearer concurrent-token"
+
+    # De-duplication must be preserved: three concurrent callers racing for
+    # the same (initially uncached) server must still only perform ONE real
+    # token fetch, not one per caller.
+    assert len(post_calls) == 1
+
+
+def test_get_authorization_header_cancelled_while_waiting_does_not_leak_lock(monkeypatch):
+    """A caller cancelled while waiting on the per-server lock must not leak it.
+
+    ``get_authorization_header`` runs ``lock.acquire()`` on a real OS thread via
+    ``asyncio.to_thread`` so a blocking wait never blocks the event loop. Once that
+    thread has actually started running ``lock.acquire()``, Python cannot interrupt
+    it: cancelling the *caller* only stops the caller from continuing, it does not
+    stop the thread. If cancellation at that await let the thread go on to acquire
+    the lock unobserved (nobody left holding a reference that will call
+    ``release()`` for it), the lock would stay held forever and every subsequent
+    call for this server would block permanently at the same line -- the very
+    cross-thread deadlock this file's lock was introduced to fix, reintroduced via
+    a different path.
+
+    This test holds the per-server lock (simulating another in-flight caller),
+    starts a second caller that has to wait for it, cancels that waiter while it
+    is genuinely blocked in its executor thread, releases the original holder, and
+    then asserts a third caller completes within a bounded timeout and performs
+    exactly one token fetch. Every potentially-hanging await is wrapped in a
+    bounded timeout so a regression fails this test quickly instead of hanging the
+    suite.
+    """
+    post_calls: list[dict[str, Any]] = []
+
+    def _client_factory(*args, **kwargs):
+        return _MockAsyncClient(
+            payload={
+                "access_token": "after-cancel-token",
+                "token_type": "Bearer",
+                "expires_in": 3600,
+            },
+            post_calls=post_calls,
+            **kwargs,
+        )
+
+    monkeypatch.setattr("httpx.AsyncClient", _client_factory)
+
+    config = ExtensionsConfig.model_validate(
+        {
+            "mcpServers": {
+                "secure-http": {
+                    "enabled": True,
+                    "type": "http",
+                    "url": "https://api.example.com/mcp",
+                    "oauth": {
+                        "enabled": True,
+                        "token_url": "https://auth.example.com/oauth/token",
+                        "grant_type": "client_credentials",
+                        "client_id": "client-id",
+                        "client_secret": "client-secret",
+                    },
+                }
+            }
+        }
+    )
+
+    manager = OAuthTokenManager.from_extensions_config(config)
+    lock = manager._locks["secure-http"]
+
+    async def scenario() -> None:
+        # Simulate another in-flight caller already holding the per-server lock
+        # (uncontended, so this succeeds immediately without blocking).
+        lock.acquire()
+        try:
+            waiter = asyncio.create_task(manager.get_authorization_header("secure-http"))
+
+            # Let the waiter's asyncio.to_thread(lock.acquire) actually get
+            # scheduled onto an executor thread and start genuinely blocking on
+            # the real lock before cancelling it -- otherwise the cancellation
+            # could land before the thread even starts, which would not exercise
+            # the bug.
+            await asyncio.sleep(0.2)
+
+            waiter.cancel()
+            # The original holder finishes its own work and releases *before* we
+            # wait on the cancelled waiter: a correct fix must keep the lock's
+            # eventual acquisition shielded from this coroutine's cancellation and
+            # wait for it to actually land before releasing, so awaiting the
+            # cancelled waiter can legitimately block until the lock is free
+            # either way.
+            lock.release()
+
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(waiter, timeout=5)
+
+            # The crux of the regression: under the bug, the waiter's abandoned
+            # executor thread went on to acquire the lock with nobody left to
+            # release it, so this third call would block forever. Bound it so a
+            # regression fails fast instead of hanging the test itself.
+            third = await asyncio.wait_for(manager.get_authorization_header("secure-http"), timeout=5)
+            assert third == "Bearer after-cancel-token"
+        finally:
+            # Test-only safety net, independent of the assertions above: under
+            # the bug, the lock is left permanently locked with a background
+            # thread (from whichever caller's orphaned acquisition landed last)
+            # still parked on a *subsequent* acquire() that will now never
+            # return. asyncio.run()'s own teardown joins every thread the
+            # default executor ever created before it returns, so leaving that
+            # thread stuck would hang this test process at interpreter/loop
+            # shutdown even after the failure above is already reported. Forcing
+            # the lock open here lets any such thread finish so the process can
+            # exit; it is a no-op once the fix keeps the lock correctly balanced.
+            if lock.locked():
+                lock.release()
+
+    asyncio.run(scenario())
+
+    # Exactly one real token fetch: the cancelled waiter must never reach
+    # _fetch_token, so the third call is the only one that performs it.
+    assert len(post_calls) == 1