fix(mcp): enforce session pool capacity during promotion (#4962)

* fix(mcp): enforce session pool capacity during promotion

* test(mcp): cover concurrent session promotion

* docs(mcp): document promotion-time capacity check

* fix(mcp): align capacity eviction with owner promotion

* fix(mcp): detach promotion eviction teardown from new owner

* test(mcp): keep eviction teardown regression focused

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
Co-authored-by: XIIRUAN <253657638+XIIRUAN@users.noreply.github.com>

backend/packages/harness/deerflow/mcp/AGENTS.md

@@ -6,6 +6,7 @@
 - **Long-running ordinary task driver**: `extensions_config.json -> mcpServers.<server>.task_toolsets` binds exact raw submit/status/cancel names; one raw tool may occupy only one role across that server's groups. `mcp/tools.py` hides status/cancel and replaces submit with a wrapper that returns only the local task ID after persistence. `ordinary.py` reads only MCP `structuredContent`, maps remote `running` to `working`, and treats `error_code=task_not_found` or malformed structured output as permanent failure. A status call with `isError=true` is a retryable call failure: the first text content block is retained as a bounded diagnostic, while a permanent remote-task outcome must arrive in a normal result with structured `status=failed`. `task_tool_caller.py` restores the same `(server_name, user_id:thread_id)` stdio session scope; HTTP/SSE calls remain ephemeral, apply `session_init_timeout` to initialization and `tool_call_timeout` to task calls, and support server-level OAuth refresh outside an Agent run. `McpTaskService` exponentially backs off transient status/cancel errors without a maximum attempt count, derives API `tracking_degraded` from the consecutive-error threshold, keeps `input_required` on a slower poll, and caps finite positive remote poll hints at 24 hours. Task-enabled server runtime/binding configuration and `mcpInterceptors` are frozen to the Gateway startup snapshot; hot drift fails clearly before tool discovery can diverge from background calls, while presentation-only fields and non-task servers remain reloadable. Configured task toolsets fail startup when the runtime is disabled or persistence is memory. Users still cannot submit an answer back to an `input_required` remote task.
 - **Durable task payload bounds**: persisted task errors are capped at 4,000 characters. `input_required` and `result_artifact` must each serialize as valid JSON within 64 KiB; an invalid or oversized payload becomes a permanent protocol failure rather than being truncated and changing its semantics. Remote task IDs/task names are limited to 255 characters and task-enabled server names to 128, matching the SQL schema; an oversized submitted remote ID is rejected only after the Service has the handle so compensation cancellation still runs. Oversized results retain the existing bounded preview/truncation/artifact behavior.
 - **Lazy initialization**: Tools loaded on first use via `get_cached_mcp_tools()`
+- **Persistent stdio session capacity**: `MCPSessionPool.MAX_SESSIONS` is a hard cap on the live LRU registry. Capacity is enforced both before creating a session and when an in-flight session is promoted, because different keys can finish initialization concurrently after all observing spare capacity. Promotion-time victims are signalled and drained by separately tracked teardown tasks on their owning loops; the newly promoted owner must never await a victim, so a blocked victim exit cannot prevent its own closure or disconnect recovery.
 - **Cache invalidation**: Detects extensions-config changes by comparing the resolved config path and a `(mtime, size, sha256)` content signature against the values recorded at initialization, not a strict mtime `>` comparison. This catches same-second edits, mtime that stays put or moves backward (`git checkout`, `cp -p` / backup restore, `tar` / `rsync`, object-store / network mounts), and a switch to a different config file with an equal-or-older mtime. The signature helper (`config/file_signature.py::get_config_signature`) is shared with `config/app_config.py::get_app_config()` for the sibling runtime-editable config file, rather than each maintaining its own copy. `ExtensionsConfig.resolve_config_path()` raises `FileNotFoundError` for an explicit `config_path`/`DEER_FLOW_EXTENSIONS_CONFIG_PATH` that points at a missing file — an operator-asserted path going missing is a real misconfiguration, so this is intentionally loud for callers that load the config for actual use (e.g. `from_file()` via `get_mcp_tools()`); only the fallback search mode returns `None`. The MCP cache's own path resolution (`mcp/cache.py::_resolve_config_path`) is narrower: it catches that specific `FileNotFoundError` locally and treats it the same as "unconfigured", so this staleness check degrades to "not stale" instead of propagating an exception when a previously-valid explicit/env-var config disappears mid-run. If `initialize_mcp_tools()` itself observes a config-signature change between the pre-load and post-load snapshots, that discard branch must reset the tool cache through the same session-pool retirement path as normal stale invalidation before waiters retry; otherwise a stale load can leave `(server_name, scope_key)` sessions from the abandoned connection available to the next wrapper build.
 - **Transports**: stdio (command-based), SSE, HTTP
 - **Per-server tool-name prefixing**: `mcpServers.<server>.tool_name_prefix` defaults to `true`, preserving the collision-safe `<server_name>_` prefix. Servers whose tools already carry a stable namespace may set it to `false`; discovery then calls `langchain_mcp_adapters.tools.load_mcp_tools` with that server's flag. Source routing and stdio session-pool wrapping are based on the producing server and transport, never on whether the visible tool name starts with the server prefix.

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

@@ -211,14 +211,35 @@ class MCPSessionPool:
             # up holding an unmanaged session.
             loop = asyncio.get_running_loop()
             task = asyncio.current_task()
+            promoted_evicted: list[tuple[asyncio.AbstractEventLoop, asyncio.Task[Any], asyncio.Event]] = []
             with self._lock:
                 still_ours = self._inflight.get(key) == (loop, ready, task, close_evt)
                 if still_ours:
                     self._inflight.pop(key)
+                    # Different keys can finish initialization concurrently.
+                    # They may all pass the earlier capacity check while no
+                    # session is registered, so enforce the cap again at the
+                    # single owner-controlled commit point.
+                    while len(self._entries) >= self.MAX_SESSIONS:
+                        oldest_key, (_, ent_loop, ent_task, ent_close) = next(iter(self._entries.items()))
+                        self._entries.pop(oldest_key)
+                        promoted_evicted.append((ent_loop, ent_task, ent_close))
                     self._entries[key] = (session, loop, task, close_evt)
                     if not ready.done():
                         ready.set_result(session)
             if still_ours:
+                # Drain victims independently: a blocked victim's __aexit__
+                # must not prevent this owner from handling its own close.
+                for ent_loop, _ent_task, ent_close in promoted_evicted:
+                    self._signal_close(ent_loop, ent_close)
+                for ent_loop, ent_task, ent_close in promoted_evicted:
+                    if ent_loop is loop:
+                        self._track_owner_teardown(ent_task)
+                    elif not ent_loop.is_closed():
+                        try:
+                            ent_loop.call_soon_threadsafe(self._track_owner_teardown, ent_task)
+                        except RuntimeError:
+                            pass  # The owning loop closed before scheduling.
                 logger.info("Created persistent MCP session for %s/%s", key[0], key[1])
             elif not ready.done():
                 ready.set_exception(asyncio.CancelledError("MCP session pool was closed while the session was being created"))
@@ -402,8 +423,8 @@ class MCPSessionPool:
                         self._inflight.pop(key)
             raise
 
-        # Phase 4: the commit inside the owner task already promoted the
-        # creation into a registered entry; nothing left to decide here.
+        # Phase 4: the owner task already promoted the initialized session and
+        # enforced capacity in the same commit critical section.
         return session
 
     # ------------------------------------------------------------------
@@ -478,7 +499,7 @@ class MCPSessionPool:
             pass
 
     def _track_owner_teardown(self, task: asyncio.Task[Any]) -> None:
-        """Keep an owner's teardown observable after its awaiter was cancelled.
+        """Keep detached eviction or cancelled-caller teardown observable.
 
         The awaiter unwinds, but the owner still has to finish ``__aexit__`` in
         its own task; the reaper awaits that completion so exceptions are
@@ -493,7 +514,7 @@ class MCPSessionPool:
             try:
                 await task
             except BaseException:
-                logger.debug("Owner task ended after caller cancellation", exc_info=True)
+                logger.debug("Owner task ended during detached teardown", exc_info=True)
 
         try:
             reaper = asyncio.get_running_loop().create_task(_reap(), name=f"mcp-session-owner-reap:{task.get_name()}")

backend/tests/test_mcp_session_pool.py

@@ -131,6 +131,134 @@ async def test_lru_eviction():
     assert cms[2].closed is False
 
 
+@pytest.mark.asyncio
+async def test_concurrent_distinct_sessions_respect_capacity():
+    """Concurrent initializations must not permanently exceed the pool cap."""
+    pool = MCPSessionPool()
+    pool.MAX_SESSIONS = 1
+    initialize_gate = asyncio.Event()
+    both_initializing = asyncio.Event()
+    initialize_count = 0
+
+    class CmFactory:
+        def __init__(self):
+            self.closed = False
+
+        async def __aenter__(self):
+            return self
+
+        async def initialize(self):
+            nonlocal initialize_count
+            initialize_count += 1
+            if initialize_count == 2:
+                both_initializing.set()
+            await initialize_gate.wait()
+
+        async def __aexit__(self, *args):
+            self.closed = True
+            return False
+
+    cms: list[CmFactory] = []
+
+    def make_cm(*_args, **_kwargs):
+        cm = CmFactory()
+        cms.append(cm)
+        return cm
+
+    with patch("langchain_mcp_adapters.sessions.create_session", side_effect=make_cm):
+        connection = {"transport": "stdio", "command": "x", "args": []}
+        first = asyncio.create_task(pool.get_session("s", "t1", connection))
+        second = asyncio.create_task(pool.get_session("s", "t2", connection))
+        await asyncio.wait_for(both_initializing.wait(), timeout=1)
+        assert len(pool._entries) == 0
+        assert len(pool._inflight) == 2
+        initialize_gate.set()
+        await asyncio.gather(first, second)
+
+    try:
+        assert len(cms) == 2
+        assert len(pool._entries) == pool.MAX_SESSIONS
+        assert len(pool._inflight) == 0
+        assert sum(cm.closed for cm in cms) == 1
+    finally:
+        await pool.close_all()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("close_mode", ["current", "disconnect", "all"])
+async def test_promoted_session_closes_while_evicted_owner_is_blocked(close_mode):
+    """An eviction victim must not hold the replacement's shutdown hostage."""
+    pool = MCPSessionPool()
+    pool.MAX_SESSIONS = 1
+    started = [asyncio.Event(), asyncio.Event()]
+    initialize = [asyncio.Event(), asyncio.Event()]
+    exiting = [asyncio.Event(), asyncio.Event()]
+    release_victim = asyncio.Event()
+    owners = []
+
+    class Session:
+        def __init__(self, index):
+            self.index = index
+
+        async def __aenter__(self):
+            self.owner = asyncio.current_task()
+            owners.append(self.owner)
+            return self
+
+        async def initialize(self):
+            started[self.index].set()
+            await initialize[self.index].wait()
+
+        async def call_tool(self, *args, **kwargs):
+            raise anyio.EndOfStream
+
+        async def __aexit__(self, *args):
+            assert asyncio.current_task() is self.owner
+            exiting[self.index].set()
+            if self.index == 0:
+                await release_victim.wait()
+
+    sessions = [Session(0), Session(1)]
+    close_task = None
+    connection = {"transport": "stdio", "command": "unused", "args": []}
+    with patch("langchain_mcp_adapters.sessions.create_session", side_effect=sessions):
+        first = asyncio.create_task(pool.get_session("s", "a", connection))
+        second = asyncio.create_task(pool.get_session("s", "b", connection))
+        try:
+            await asyncio.wait_for(asyncio.gather(*(event.wait() for event in started)), 2)
+            initialize[0].set()
+            await asyncio.wait_for(asyncio.shield(first), 2)
+            initialize[1].set()
+            replacement = await asyncio.wait_for(asyncio.shield(second), 2)
+            await asyncio.wait_for(exiting[0].wait(), 2)
+            assert len(pool._entries) == 1
+            if close_mode == "current":
+                close_task = asyncio.create_task(pool.close_session_if_current("s", "b", replacement))
+            elif close_mode == "disconnect":
+                close_task = asyncio.create_task(call_pooled_session_tool(replacement, pool, server_name="s", scope_key="b", tool_name="test", arguments={}, call_kwargs={}))
+            else:
+                close_task = asyncio.create_task(pool.close_all())
+            await asyncio.wait_for(exiting[1].wait(), 2)
+            if close_mode == "disconnect":
+                with pytest.raises(anyio.EndOfStream):
+                    await asyncio.wait_for(asyncio.shield(close_task), 2)
+            else:
+                await asyncio.wait_for(asyncio.shield(close_task), 2)
+            assert not owners[0].done()
+            assert pool._teardown_tasks
+        finally:
+            release_victim.set()
+            for event in initialize:
+                event.set()
+            await asyncio.gather(first, second, return_exceptions=True)
+            await pool.close_all()
+            if close_task is not None:
+                await asyncio.gather(close_task, return_exceptions=True)
+            await asyncio.gather(*owners, return_exceptions=True)
+            await asyncio.gather(*list(pool._teardown_tasks), return_exceptions=True)
+        assert not pool._teardown_tasks
+
+
 @pytest.mark.asyncio
 async def test_close_scope():
     """close_scope shuts down sessions for a specific scope key."""