diff --git a/CHANGELOG.md b/CHANGELOG.md index a38de09128..5e23ef542f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -965,6 +965,13 @@ This release closes that milestone with **765 merged pull requests**. the API had reported as paused. The read now takes the writer first, as every other mutating path in that repository does. PostgreSQL was unaffected. ([#5777]) +- **mcp:** Lazy MCP initialization no longer runs tool discovery twice when + discovery itself raises a `RuntimeError` such as `McpTaskConfigurationError`. + The `asyncio.run` fallback in `get_cached_mcp_tools()` was meant only for + `get_event_loop()` failing, but it also caught discovery errors and + re-spawned every stdio server (and re-fetched OAuth tokens) before giving + up; inside a running loop it also logged a misleading "asyncio.run() cannot + be called from a running event loop" traceback instead of the real cause. - **uploads:** Deleting an uploaded document no longer deletes the converted Markdown beside it. Conversion names a companion after the document's stem and falls back to a `_N` suffix when that name is taken, so the `.md` next to diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 5299e3559e..2571686c02 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -732,6 +732,12 @@ SQLite 的写锁，因此过期的读取会通过校验，并把任务状态写回 `enabled` 且不改动 `next_run_at`， 导致接口已回复“已暂停”的任务仍被继续触发。现在该读取会像该仓储中其他写入路径一样先获取写锁。 PostgreSQL 不受影响。([#5777]) +- **mcp：** MCP 延迟初始化在工具发现本身抛出 `RuntimeError`（例如 + `McpTaskConfigurationError`）时，不再把发现流程跑两遍。`get_cached_mcp_tools()` 里的 + `asyncio.run` 兜底只为 `get_event_loop()` 失败而设，却同时捕获了发现阶段的错误，于是在 + 放弃之前会重新拉起每一个 stdio 服务器（并重新获取 OAuth 令牌）；在运行中的事件循环里， + 它记录的还是误导性的 "asyncio.run() cannot be called from a running event loop" + 堆栈，而不是真正的原因。 - **上传：** 删除已上传的文档时，不再连带删除其旁边转换生成的 Markdown。转换以文档主干名 命名配套文件，名称被占用时回退为 `_N` 后缀，因此文档旁的 `.md` 可能属于主干名相同的另一个 文档，或属于用户自己：上传 `a.docx` 与 `a.pdf` 会生成 `a.md` 与 `a_1.md`，删除 `a.pdf` diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index dc691794e2..b8de900dc6 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -6,7 +6,10 @@ - **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 canonical `(server_name, versioned user/thread/incarnation scope)` stdio session as ordinary calls; non-NULL incarnations use an unambiguous versioned encoding, while an explicitly captured legacy NULL retains the pre-activation `user_id:thread_id` scope. A runtime with a missing/invalid incarnation key fails closed. 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. - **Remote task bring-up**: One `session_init_timeout` deadline covers transport context entry and MCP initialization. Report the server and configured bound on expiry; disable it before calling the tool. Enter and exit the adapter's AnyIO context managers in the same task; tests include real SSE endpoint/initialization stalls and connection cleanup. -- **Lazy initialization**: Tools loaded on first use via `get_cached_mcp_tools()` +- **Lazy initialization**: Tools loaded on first use via `get_cached_mcp_tools()`. Its + `asyncio.run` fallback covers only `get_event_loop()` raising (no loop set, or a closed + loop); a `RuntimeError` from discovery itself (e.g. `McpTaskConfigurationError`) is + logged once and never triggers a second discovery pass. - **Loop-isolated stdio sessions**: Both the live registry and in-flight creations are keyed by `(server_name, scope_key, owning_loop)`. Same-loop callers share initialization and state; live sibling loops must never cancel or replace each diff --git a/backend/packages/harness/deerflow/mcp/cache.py b/backend/packages/harness/deerflow/mcp/cache.py index d05b532e68..97d6fed0d4 100644 --- a/backend/packages/harness/deerflow/mcp/cache.py +++ b/backend/packages/harness/deerflow/mcp/cache.py @@ -381,9 +381,18 @@ def get_cached_mcp_tools() -> list[BaseTool]: retired_pool.close_all_sync() logger.info("MCP tools not initialized, performing lazy initialization...") + # Only ``get_event_loop()`` may fall back to ``asyncio.run``: a + # ``RuntimeError`` raised *by* ``initialize_mcp_tools()`` (for example + # ``McpTaskConfigurationError``) must not trigger a second discovery + # pass that respawns every stdio server and re-fetches OAuth tokens. try: - loop = asyncio.get_event_loop() - if loop.is_running(): + loop: asyncio.AbstractEventLoop | None = asyncio.get_event_loop() + except RuntimeError: + loop = None + try: + if loop is None or loop.is_closed(): + asyncio.run(initialize_mcp_tools()) + elif loop.is_running(): import concurrent.futures with concurrent.futures.ThreadPoolExecutor() as executor: @@ -391,12 +400,6 @@ def get_cached_mcp_tools() -> list[BaseTool]: future.result() else: loop.run_until_complete(initialize_mcp_tools()) - except RuntimeError: - try: - asyncio.run(initialize_mcp_tools()) - except Exception: - logger.exception("Failed to lazy-initialize MCP tools") - return [] except Exception: logger.exception("Failed to lazy-initialize MCP tools") return [] diff --git a/backend/tests/test_mcp_cache.py b/backend/tests/test_mcp_cache.py index a6580af01e..1ed11e655b 100644 --- a/backend/tests/test_mcp_cache.py +++ b/backend/tests/test_mcp_cache.py @@ -24,6 +24,7 @@ import asyncio import json import logging import os +import sys import threading from pathlib import Path @@ -1206,3 +1207,117 @@ def test_unverifiable_signature_is_not_treated_as_stable(cache_globals, monkeypa assert cache_module._read_stable_mcp_snapshot(cfg, incomplete) is None assert cache_module._is_cache_stale() is True + + +class TestLazyInitializationFailure: + """``get_cached_mcp_tools()`` must run discovery exactly once per call. + + The ``except RuntimeError`` fallback exists for ``asyncio.get_event_loop()`` + raising when no loop is set (non-main thread, or Python 3.14+). It must not + also catch a ``RuntimeError`` raised *by* ``initialize_mcp_tools()`` itself — + ``McpTaskConfigurationError`` is one, and it is deliberately re-raised by + ``get_mcp_tools`` — because the fallback then re-runs a full discovery pass + (respawning every stdio server, re-fetching OAuth tokens) before giving up. + """ + + @staticmethod + def _install_failing_discovery(monkeypatch, tmp_path: Path) -> list[str]: + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + calls: list[str] = [] + + async def _fake_tools(**_kwargs): + calls.append(threading.current_thread().name) + raise RuntimeError("discovery failed") + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + return calls + + def test_runtime_error_from_discovery_without_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog): + calls = self._install_failing_discovery(monkeypatch, tmp_path) + + with caplog.at_level("ERROR", logger="deerflow.mcp.cache"): + assert cache_module.get_cached_mcp_tools() == [] + + assert calls == [threading.current_thread().name] + assert cache_module._cache_initialized is False + logged = [record for record in caplog.records if record.exc_info] + assert len(logged) == 1 + assert str(logged[0].exc_info[1]) == "discovery failed" + + def test_runtime_error_from_discovery_inside_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog): + calls = self._install_failing_discovery(monkeypatch, tmp_path) + + async def _call_from_running_loop(): + return cache_module.get_cached_mcp_tools() + + with caplog.at_level("ERROR", logger="deerflow.mcp.cache"): + assert asyncio.run(_call_from_running_loop()) == [] + + assert len(calls) == 1 + assert cache_module._cache_initialized is False + # The logged failure must be the discovery error, not a misleading + # "asyncio.run() cannot be called from a running event loop". + logged = [record for record in caplog.records if record.exc_info] + assert len(logged) == 1 + assert str(logged[0].exc_info[1]) == "discovery failed" + + def test_thread_without_event_loop_still_initializes_once(self, cache_globals, monkeypatch, tmp_path): + """The no-loop fallback stays intact: a worker thread with no loop set initializes via ``asyncio.run``.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + calls: list[str] = [] + + async def _fake_tools(**_kwargs): + calls.append(threading.current_thread().name) + return [] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + outcome: dict[str, object] = {} + + def _worker(): + # ``Thread.join`` does not propagate exceptions: capture them so a + # regression surfaces its real traceback instead of a bare KeyError. + try: + with pytest.raises(RuntimeError): + asyncio.get_event_loop() + outcome["result"] = cache_module.get_cached_mcp_tools() + except BaseException: + outcome["error"] = sys.exc_info() + + worker = threading.Thread(target=_worker, name="mcp-no-loop-worker") + worker.start() + worker.join(timeout=10) + assert not worker.is_alive() + if "error" in outcome: + _, exc, tb = outcome["error"] + raise exc.with_traceback(tb) + + assert outcome["result"] == [] + assert len(calls) == 1 + assert cache_module._cache_initialized is True + + def test_closed_current_loop_falls_back_to_fresh_loop(self, cache_globals, monkeypatch, tmp_path): + """A closed loop left as the current loop must not be driven; fall back to ``asyncio.run``.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + calls: list[str] = [] + + async def _fake_tools(**_kwargs): + calls.append(threading.current_thread().name) + return [] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + closed_loop = asyncio.new_event_loop() + closed_loop.close() + asyncio.set_event_loop(closed_loop) + try: + assert cache_module.get_cached_mcp_tools() == [] + finally: + asyncio.set_event_loop(None) + + assert len(calls) == 1 + assert cache_module._cache_initialized is True