fix(db): 路由跨 loop 的 DB 访问回引擎主 loop,消除 asyncmy "attached to a different loop"

会商/定时任务/沙箱工具等在 worker 线程的临时 loop 上用 asyncio.run 跑全局引擎连接池的 DB 操作,连接被绑定到随即关闭的临时 loop;池子回收/释放该连接时 asyncmy _terminate_graceful_close 在错误的 loop 上触发 future,持续刷 "got Future attached to a different loop"。 - engine.py: 新增 run_db_blocking(fire_and_forget_db_write 的带返回值版本)。离开主 loop 时用 run_coroutine_threadsafe 把协程送回主 loop 执行、阻塞调用线程拿结果;仅在无主 loop 或正同步处于主 loop 内(送回会死锁)时退回临时 loop。 - thread_paths / prompt / skill_tools: 三处同步→异步阻塞读桥(任务owner解析/常驻技能/ 技能可见性)改走 run_db_blocking,删除各自重复的 _run_coro_blocking。 - llm/tool metrics 中间件: fire-and-forget 写改走 fire_and_forget_db_write。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-24 15:57:04 +08:00
1 parent 5ecf28e60c
commit 6aff165d09
@@ -86,27 +86,6 @@ def _drop_archived_skills(skills: list[Skill]) -> list[Skill]:
        return skills


-def _run_coro_blocking(coro):
-    """Run an async coroutine to completion from sync code.
-
-    Mirrors ``skill_tools._run_coro_blocking``: when already inside a running
-    event loop (e.g. prompt built during an async agent run), hand the coroutine
-    to a fresh loop in a worker thread; otherwise run it directly.
-    """
-    try:
-        asyncio.get_running_loop()
-        in_loop = True
-    except RuntimeError:
-        in_loop = False
-
-    if in_loop:
-        import concurrent.futures
-
-        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-            return pool.submit(asyncio.run, coro).result()
-    return asyncio.run(coro)
-
-
 def _load_always_on_names() -> set[str]:
    """Names marked 常驻(免压缩), read from the **DB** (skills table).

@@ -114,11 +93,11 @@ def _load_always_on_names() -> set[str]:
    simply treats every skill as compressible (safe default).
    """
    try:
-        from deerflow.persistence.engine import get_session_factory
+        from deerflow.persistence.engine import get_session_factory, run_db_blocking
        from deerflow.persistence.skills import make_skill_store

        store = make_skill_store(get_session_factory())
-        return set(_run_coro_blocking(store.list_always_on_names()))
+        return set(run_db_blocking(store.list_always_on_names()))
    except Exception:
        logger.debug("Failed to load always_on skill names from DB", exc_info=True)
        return set()
@@ -20,7 +20,6 @@ not the retry-wrapped successful response.

 from __future__ import annotations

-import asyncio
 import logging
 import time
 import uuid
@@ -38,7 +37,7 @@ from langchain.agents.middleware.types import (
 from langchain_core.messages import AIMessage
 from langgraph.errors import GraphBubbleUp

-from deerflow.persistence.engine import get_session_factory
+from deerflow.persistence.engine import fire_and_forget_db_write, get_session_factory
 from deerflow.persistence.llm_metrics import LlmMetricRecord, make_llm_metrics_store
 from deerflow.persistence.llm_metrics.base import LlmMetricsStore
 from deerflow.runtime.user_context import get_effective_user_id
@@ -280,16 +279,13 @@ class LlmMetricsMiddleware(AgentMiddleware[AgentState]):
        store = self._store()
        if store is None:
            return
+        # Route the best-effort write onto the engine's home loop. Doing
+        # ``asyncio.run(store.record(...))`` here (when the model call ran in a worker
+        # thread) would reuse the global engine's pooled connections on a fresh loop and
+        # raise "got Future attached to a different loop" — harmless (caught below) but
+        # log-spammy and pool-churning under concurrency. See ``fire_and_forget_db_write``.
        try:
-            try:
-                loop = asyncio.get_running_loop()
-            except RuntimeError:
-                loop = None
-            if loop is not None and loop.is_running():
-                # Fire-and-forget — never let metric IO block the request path.
-                loop.create_task(store.record(record))
-            else:
-                asyncio.run(store.record(record))
+            fire_and_forget_db_write(store.record(record))
        except Exception:
            logger.warning("LlmMetricsMiddleware: failed to flush metric %s", record.id, exc_info=True)

@@ -25,7 +25,6 @@ agent runtime keeps running.

 from __future__ import annotations

-import asyncio
 import logging
 import re
 import time
@@ -41,7 +40,7 @@ from langgraph.errors import GraphBubbleUp
 from langgraph.prebuilt.tool_node import ToolCallRequest
 from langgraph.types import Command

-from deerflow.persistence.engine import get_session_factory
+from deerflow.persistence.engine import fire_and_forget_db_write, get_session_factory
 from deerflow.persistence.tool_metrics import ToolMetricRecord, make_tool_metrics_store
 from deerflow.persistence.tool_metrics.base import ToolMetricsStore
 from deerflow.runtime.user_context import get_effective_user_id
@@ -262,16 +261,12 @@ class ToolMetricsMiddleware(AgentMiddleware[AgentState]):
        store = self._store()
        if store is None:
            return
+        # Route the best-effort write onto the engine's home loop (a tool call that ran
+        # in a worker thread would otherwise do ``asyncio.run(store.record(...))`` and reuse
+        # the global engine's pooled connections cross-loop → "got Future attached to a
+        # different loop"). See ``fire_and_forget_db_write``.
        try:
-            try:
-                loop = asyncio.get_running_loop()
-            except RuntimeError:
-                loop = None
-            if loop is not None and loop.is_running():
-                # Fire-and-forget — never let metric IO block the tool path.
-                loop.create_task(store.record(record))
-            else:
-                asyncio.run(store.record(record))
+            fire_and_forget_db_write(store.record(record))
        except Exception:
            logger.warning("ToolMetricsMiddleware: failed to flush metric %s", record.id, exc_info=True)

@@ -32,6 +32,13 @@ logger = logging.getLogger(__name__)

 _engine: AsyncEngine | None = None
 _session_factory: async_sessionmaker[AsyncSession] | None = None
+# The event loop on which the async engine was created. asyncmy/asyncpg connection
+# pools bind their connections to the loop that opened them; awaiting a write on a
+# *different* loop reuses those connections cross-loop and raises
+# "got Future attached to a different loop". We capture the engine's home loop here so
+# best-effort writes coming from worker threads / other loops can be routed back to it
+# (see ``fire_and_forget_db_write``).
+_engine_loop: asyncio.AbstractEventLoop | None = None


 def _add_runtime_vendor_path() -> None:
@@ -293,12 +300,20 @@ async def init_engine(
        pool_timeout: Seconds to wait for a free pooled connection before erroring (postgres/mysql).
        sqlite_dir: Directory to create for SQLite (ensured to exist).
    """
-    global _engine, _session_factory
+    global _engine, _session_factory, _engine_loop

    if backend == "memory":
        logger.info("Persistence backend=memory -- ORM engine not initialized")
        return

+    # Remember the loop that owns the engine's connection pool, so off-loop
+    # best-effort writes (metrics from worker threads, etc.) can be routed here
+    # instead of reusing pooled connections cross-loop.
+    try:
+        _engine_loop = asyncio.get_running_loop()
+    except RuntimeError:
+        _engine_loop = None
+
    if backend == "postgres":
        try:
            import asyncpg  # noqa: F401
@@ -459,11 +474,121 @@ def get_engine() -> AsyncEngine | None:
    return _engine


+def get_engine_loop() -> asyncio.AbstractEventLoop | None:
+    """Return the event loop that owns the engine's connection pool (or None)."""
+    return _engine_loop
+
+
+def fire_and_forget_db_write(coro) -> None:
+    """Schedule a best-effort DB write on the engine's home loop, from any loop/thread.
+
+    The async engine's connection pool is bound to the loop that created it
+    (``init_engine``, normally the Gateway's main loop). Running a write on a
+    *different* loop — e.g. a model/tool call that LangGraph executed in a worker
+    thread, which then does ``asyncio.run(store.record(...))`` — reuses those pooled
+    connections cross-loop and raises ``got Future attached to a different loop``.
+    Under concurrency this happens constantly (many calls land in worker threads at
+    once), spamming the log and churning the pool; single-threaded it almost never does.
+
+    This routes every such write back onto the engine's home loop:
+    - already on that loop  -> ``create_task`` (in-loop fire-and-forget);
+    - on another loop/thread -> ``run_coroutine_threadsafe`` to the home loop.
+
+    The write itself stays best-effort: if no engine loop is captured (memory backend
+    / not initialized) the coroutine is closed (avoids a "never awaited" warning) and
+    the call is a no-op. Never raises.
+    """
+    loop = _engine_loop
+    if loop is None or loop.is_closed():
+        close = getattr(coro, "close", None)
+        if callable(close):
+            close()
+        return
+    try:
+        running = asyncio.get_running_loop()
+    except RuntimeError:
+        running = None
+    try:
+        if running is loop:
+            loop.create_task(coro)
+        else:
+            asyncio.run_coroutine_threadsafe(coro, loop)
+    except Exception:  # noqa: BLE001 — best-effort; scheduling must never break the caller
+        close = getattr(coro, "close", None)
+        if callable(close):
+            close()
+        logger.debug("fire_and_forget_db_write: failed to schedule DB write", exc_info=True)
+
+
+def _run_coro_on_throwaway_loop(coro):
+    """Last-resort: run *coro* to completion on a fresh event loop.
+
+    Only reached when there is no usable engine home loop, or when the caller is
+    *synchronously* inside the home loop (routing back would deadlock). This
+    touches the shared connection pool from a throwaway loop, so it can still leak
+    a dead-loop-bound connection — but the common worker-thread path no longer
+    gets here (see ``run_db_blocking``).
+    """
+    try:
+        asyncio.get_running_loop()
+        in_loop = True
+    except RuntimeError:
+        in_loop = False
+
+    if in_loop:
+        import concurrent.futures
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+            return pool.submit(asyncio.run, coro).result()
+    return asyncio.run(coro)
+
+
+def run_db_blocking(coro, *, timeout: float | None = 30.0):
+    """Run an awaited DB coroutine from sync code, safe w.r.t. the engine's home loop.
+
+    The blocking-read sibling of :func:`fire_and_forget_db_write`: where that one
+    drops the result, this returns it — used by the sync→async bridges that must
+    read a row before proceeding (thread-path owner resolution, always-on skills,
+    skill visibility).
+
+    The async engine's connection pool is bound to the loop that created it
+    (``init_engine``). Running a DB coroutine on a *throwaway* loop
+    (``asyncio.run``) checks out pooled connections cross-loop and **leaks** them:
+    the connection ends up bound to a loop that is then closed, and when the pool
+    later recycles/disposes it, asyncmy's ``_terminate_graceful_close`` fires on the
+    wrong loop and raises ``got Future <...> attached to a different loop``. Under
+    concurrency (scheduler / subagent / sandbox-tool worker threads each spinning a
+    fresh loop) this floods the logs.
+
+    So when we are **off** the home loop — the common case, a worker thread with no
+    running loop — route the coroutine onto the home loop and block the *caller*
+    thread for its result. The pool is only ever touched on its own loop, so nothing
+    leaks. Only when there is no home loop, or we are synchronously inside it (where
+    blocking would deadlock), do we fall back to a throwaway loop.
+
+    Never swallows the coroutine's own exception — the sync bridges wrap this call in
+    their own try/except and degrade gracefully.
+    """
+    loop = _engine_loop
+    try:
+        running = asyncio.get_running_loop()
+    except RuntimeError:
+        running = None
+    if loop is not None and not loop.is_closed() and running is not loop:
+        # Off the home loop (worker thread, or a different loop): execute the coro
+        # on the engine's loop and block this thread for the result. asyncmy's
+        # pooled connections stay on their own loop, so none leak cross-loop.
+        return asyncio.run_coroutine_threadsafe(coro, loop).result(timeout)
+    # No home loop, or we're already synchronously inside it: last resort.
+    return _run_coro_on_throwaway_loop(coro)
+
+
 async def close_engine() -> None:
    """Dispose the engine, release all connections."""
-    global _engine, _session_factory
+    global _engine, _session_factory, _engine_loop
    if _engine is not None:
        await _engine.dispose()
        logger.info("Persistence engine closed")
    _engine = None
    _session_factory = None
+    _engine_loop = None
@@ -28,7 +28,6 @@ safely to the pre-fix semantics.

 from __future__ import annotations

-import asyncio
 import logging
 import threading

@@ -106,27 +105,6 @@ def _is_creator_shared_thread(metadata: dict) -> bool:
    return bool(metadata.get("taskId")) or metadata.get("thread_type") == "roundtable"


-def _run_coro_blocking(coro):
-    """Run an async coroutine to completion from sync code.
-
-    Mirrors ``lead_agent.prompt._run_coro_blocking``: when already inside a
-    running event loop, hand the coroutine to a fresh loop in a worker thread;
-    otherwise run it directly.
-    """
-    try:
-        asyncio.get_running_loop()
-        in_loop = True
-    except RuntimeError:
-        in_loop = False
-
-    if in_loop:
-        import concurrent.futures
-
-        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-            return pool.submit(asyncio.run, coro).result()
-    return asyncio.run(coro)
-
-
 def resolve_path_user_id(thread_id: str | None) -> str:
    """Canonical user bucket for *thread_id*'s files (sync).

@@ -140,7 +118,9 @@ def resolve_path_user_id(thread_id: str | None) -> str:
    if cached is not None:
        return cached
    try:
-        owner = _run_coro_blocking(_fetch_task_owner(thread_id))
+        from deerflow.persistence.engine import run_db_blocking
+
+        owner = run_db_blocking(_fetch_task_owner(thread_id))
    except Exception:
        logger.debug("Failed to resolve task-shared owner for thread %s", thread_id, exc_info=True)
        owner = None
@@ -75,30 +75,6 @@ def _resolve_effective_user_id() -> str | None:
        return None


-def _run_coro_blocking(coro):
-    """在同步工具里把一个协程跑到完成。
-
-    ``skill_list`` 是同步 ``@tool``:LangChain 通常在线程池里执行它(无运行中的
-    事件循环),但 curator 评审会在已有事件循环的线程里调用它。两种情况都要能
-    拿到 ``SkillStore`` 的异步查询结果 —— 检测到运行中的事件循环就转交独立线程
-    执行 ``asyncio.run``,否则直接跑。
-    """
-    import asyncio
-
-    try:
-        asyncio.get_running_loop()
-        in_loop = True
-    except RuntimeError:
-        in_loop = False
-
-    if in_loop:
-        import concurrent.futures
-
-        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-            return pool.submit(asyncio.run, coro).result()
-    return asyncio.run(coro)
-
-
 def _load_owned_skill_names(user_id: str | None) -> set[str] | None:
    """返回 ``owner_user_id == user_id`` 的自定义技能名集合。

@@ -110,14 +86,14 @@ def _load_owned_skill_names(user_id: str | None) -> set[str] | None:
    if not user_id:
        return None
    try:
-        from deerflow.persistence.engine import get_session_factory
+        from deerflow.persistence.engine import get_session_factory, run_db_blocking
        from deerflow.persistence.skills import make_skill_store

        factory = get_session_factory()
        if factory is None:
            return None
        store = make_skill_store(factory)
-        records = _run_coro_blocking(store.list_visible(user_id))
+        records = run_db_blocking(store.list_visible(user_id))
        return {r["name"] for r in records if r.get("owner_user_id") == user_id}
    except Exception:
        logger.debug("Failed to load owned skill names for skill_list filter", exc_info=True)