diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 030db61ff14..d152af3133e 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -6,6 +6,7 @@ - **Long-running ordinary task driver**: `extensions_config.json -> mcpServers..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..tool_name_prefix` defaults to `true`, preserving the collision-safe `_` 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. diff --git a/backend/packages/harness/deerflow/mcp/session_pool.py b/backend/packages/harness/deerflow/mcp/session_pool.py index 333505fdbec..b3d178b3c86 100644 --- a/backend/packages/harness/deerflow/mcp/session_pool.py +++ b/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 diff --git a/backend/tests/test_mcp_session_pool.py b/backend/tests/test_mcp_session_pool.py index 01fa986dc7b..a3a6f718add 100644 --- a/backend/tests/test_mcp_session_pool.py +++ b/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."""