fix(mcp): enforce session pool capacity during promotion

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.
 - **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
 - **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

@@ -251,14 +251,28 @@ async def get_session(
         # that case we must NOT resurrect the session into _entries. Instead we
         # own the teardown: signal our owner task and wait for it to run
         # __aexit__ in its own task, then surface the cancellation.
+        promoted_evicted: list[tuple[asyncio.AbstractEventLoop, asyncio.Task[Any], asyncio.Event]] = []
         with self._lock:
             still_ours = self._inflight.get(key) == (current_loop, ready, task, close_evt)
             if still_ours:
                 self._inflight.pop(key)
+                # Different keys can finish initialization concurrently. They
+                # all pass the Phase 1 capacity check while _entries is still
+                # empty, so enforce the cap again when each live session is
+                # promoted into the LRU registry.
+                while len(self._entries) >= self.MAX_SESSIONS:
+                    oldest_key, (_, loop, ent_task, ent_close) = next(iter(self._entries.items()))
+                    self._entries.pop(oldest_key)
+                    promoted_evicted.append((loop, ent_task, ent_close))
                 self._entries[key] = (session, current_loop, task, close_evt)
         if not still_ours:
             await self._shutdown(close_evt, task)
             raise asyncio.CancelledError("MCP session pool was closed while the session was being created")
+        for loop, ent_task, ent_close in promoted_evicted:
+            if loop is current_loop and not loop.is_closed():
+                await self._shutdown(ent_close, ent_task)
+            else:
+                self._signal_close(loop, ent_close)
         logger.info("Created persistent MCP session for %s/%s", server_name, scope_key)
         return session
 

backend/tests/test_mcp_session_pool.py

@@ -125,6 +125,59 @@ def make_cm(*a, **kw):
     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
 async def test_close_scope():
     """close_scope shuts down sessions for a specific scope key."""