fix: event loop closed bug

apps/application/workflow/common.py

@@ -8,8 +8,9 @@
 """
 
 import asyncio
-import contextlib
+import os
 import threading
+from concurrent.futures import CancelledError
 from enum import Enum
 from typing import List, Dict
 
@@ -257,44 +258,82 @@ class CancelledException(Exception):
     pass
 
 
+class _SharedEventLoop:
+    """
+    进程内常驻事件循环(后台守护线程)。
+
+    模型实例按 model_id 缓存,其 httpx 异步连接池随实例复用;连接绑定在创建它的事件循环上。
+    协程内的阻塞调用会卡住所有工作流,需用 asyncio.to_thread 挪出。
+    """
+
+    _loop = None
+    _pid = None
+    _lock = threading.Lock()
+
+    @classmethod
+    def get(cls):
+        loop = cls._loop
+        # gunicorn 等 fork 出的子进程不继承父进程线程,按 pid 判断是否需要重建
+        if loop is not None and cls._pid == os.getpid() and loop.is_running():
+            return loop
+        with cls._lock:
+            if cls._loop is None or cls._pid != os.getpid() or not cls._loop.is_running():
+                loop = asyncio.new_event_loop()
+                ready = threading.Event()
+
+                def run():
+                    asyncio.set_event_loop(loop)
+                    loop.call_soon(ready.set)
+                    loop.run_forever()
+
+                thread = threading.Thread(target=run, name="maxkb-workflow-event-loop", daemon=True)
+                loop._thread = thread
+                thread.start()
+                ready.wait()
+                cls._loop, cls._pid = loop, os.getpid()
+        return cls._loop
+
+
 class AsyncRunner:
     """跨线程取消感知地运行异步协程;cancel() 置位事件使阻塞中的流式中断并抛 CancelledException。"""
 
     def __init__(self):
         self.cancel_event = threading.Event()
+        self._lock = threading.Lock()
+        self._loop = None
+        self._task = None
 
     def cancel(self):
-        """请求取消(任意线程可调);只置旗标,真正中断发生在事件循环内。"""
+        """请求取消(任意线程可调):置位旗标并取消正在运行的协程。"""
         self.cancel_event.set()
+        with self._lock:
+            loop, task = self._loop, self._task
+        if loop is not None and task is not None:
+            loop.call_soon_threadsafe(task.cancel)
 
     def run_async(self, async_factory):
-        """在独立事件循环里运行 async_factory() 的协程;被取消则抛 CancelledException。"""
-        return asyncio.run(self.await_or_cancel(async_factory()))
-
-    async def _wait_cancel(self):
-        # 轮询取消旗标,置位即结束,充当竞速对手
-        while not self.cancel_event.is_set():
-            await asyncio.sleep(0)
-
-    async def await_or_cancel(self, _async):
-        """竞速等待:流式完成返回其结果;取消先到则抛 CancelledException。"""
-
-        # 真正要跑的流式协程,包成 task 才能与 waiter 一起放进 asyncio.wait
-        stream = asyncio.create_task(_async)
-        # 把"是否已取消"伪装成一个可等待的 task,充当 stream 的竞速对手。
-        waiter = asyncio.create_task(self._wait_cancel())
-        # 同时等两个 task,谁先完成(返回)谁进入 done;FIRST_COMPLETED = 取先到者。
-        done, _ = await asyncio.wait({stream, waiter}, return_when=asyncio.FIRST_COMPLETED)
-        # stream_done 为真 -> 流式自己先跑完;为假 -> 是 waiter 先完成,即取消先到。
-        stream_done = stream in done
-        # 无论谁赢,都把两个 task cancel 并等它们退干净,防止"输掉的那一方"
-        # 还悬在事件循环里。await 吞掉 CancelledError(取消是本意,不算错误)。
-        for task in (stream, waiter):
-            task.cancel()
-            with contextlib.suppress(asyncio.CancelledError):
-                await task
-
-        # 流式正常结束:把它的返回值交给调用方。
-        if stream_done:
-            return stream.result()
-        raise CancelledException()
+        """在进程内常驻事件循环中运行 async_factory() 的协程并阻塞等待结果;被取消则抛 CancelledException。"""
+        if self.cancel_event.is_set():
+            raise CancelledException()
+        loop = _SharedEventLoop.get()
+        if threading.current_thread() is loop._thread:
+            # 在常驻循环线程内同步等待自身会死锁;协程内应直接 await
+            raise RuntimeError("AsyncRunner.run_async cannot be called from the shared event loop thread")
+        future = asyncio.run_coroutine_threadsafe(self._run(async_factory), loop)
+        try:
+            # 取消时任务先在循环内执行完 finally 清理(落盘已累积内容、关闭流),再以 CancelledError 结束
+            return future.result()
+        except CancelledError:
+            raise CancelledException()
+
+    async def _run(self, async_factory):
+        with self._lock:
+            self._loop, self._task = asyncio.get_running_loop(), asyncio.current_task()
+        try:
+            # 登记任务前已请求取消:直接结束,不发起模型请求
+            if self.cancel_event.is_set():
+                raise CancelledException()
+            return await async_factory()
+        finally:
+            with self._lock:
+                self._loop, self._task = None, None

apps/application/workflow/nodes/ai_chat_node/agent.py

@@ -186,13 +186,11 @@ async def _run():
             stream_mode="messages",
         )
 
-        async def consume():
-            # runner 取消该 task 会让 CancelledError 注入在途的工具 await,从而打断正在运行的工具。
-            async for chunk in response:
-                msg = chunk[0]
-                call_back.on_next(msg)
-
-        await async_runner.await_or_cancel(consume())
+        # 取消:run_async 的 task.cancel() 会让 CancelledError 注入在途的 chunk await,
+        # 从而打断正在运行的 MCP/工具调用。
+        async for chunk in response:
+            msg = chunk[0]
+            call_back.on_next(msg)
 
     def _classify_error(e):
         # 取消：原样保留（保持节点取消语义）；MCP TaskGroup 的 ExceptionGroup：展开取真实异常并包成 RuntimeError
@@ -205,7 +203,7 @@ def _classify_error(e):
 
     error = None
     try:
-        asyncio.run(_run())
+        async_runner.run_async(_run)
     except Exception as e:
         error = _classify_error(e)
     finally:

apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py

@@ -7,6 +7,7 @@
 @desc:
 """
 
+import asyncio
 import base64
 import contextlib
 import json
@@ -473,7 +474,9 @@ async def _stream_response(
 
         # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接
         finally:
-            self._write_final_context(chat_model, message_list, question, answer, reasoning_content)
+            await asyncio.to_thread(
+                self._write_final_context, chat_model, message_list, question, answer, reasoning_content
+            )
             with contextlib.suppress(Exception):
                 await response.aclose()
 
@@ -561,7 +564,6 @@ def _handle_mcp(
         ToolExecutor().validate_mcp_transport(json.dumps(mcp_servers_config))
 
         if tools or mcp_servers_config or skill_tool_ids:
-            node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.RUNNING)
             # 使用可变状态在回调间共享（answer 累积、当前文本 content id、工具 content id 映射）
             state = {"answer": ""}
             tool_stream = ToolCallStreamManagement()

apps/application/workflow/nodes/ai_chat_node/tools/tool/__init__.py

@@ -22,5 +22,5 @@ def get_tool_tools(source_type, source_id, tool_ids, workspace_id, cancel_event,
     if not tool_ids:
         return []
     return get_workflow_tools(
-        source_type, source_id, tool_ids, workspace_id, workflow_params, cancel_event
+        source_type, source_id, tool_ids, workspace_id, cancel_event, workflow_params
     ) + get_custom_tools(source_type, source_id, tool_ids, workspace_id, cancel_event)

apps/application/workflow/nodes/ai_chat_node/tools/tool/custom.py

@@ -85,7 +85,7 @@ def _run():
     threading.Thread(target=_run, daemon=True).start()
     if await_complete_or_cancel(done_event, cancel_event, on_cancel):
         raise CancelledException()
-    if result["error"]:
+    if result.get("error"):
         raise result["error"]
     return result["value"]
 

apps/application/workflow/nodes/image_understand_node/image_understand_node.py

@@ -1,4 +1,5 @@
 # coding=utf-8
+import asyncio
 import base64
 import contextlib
 import uuid_utils.compat as uuid
@@ -258,7 +259,9 @@ async def _stream_response(
 
         # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接
         finally:
-            self._write_final_context(chat_model, message_list, question, answer, reasoning_content)
+            await asyncio.to_thread(
+                self._write_final_context, chat_model, message_list, question, answer, reasoning_content
+            )
             with contextlib.suppress(Exception):
                 await response.aclose()
 

apps/application/workflow/nodes/llm_node/llm_node.py

@@ -7,6 +7,7 @@
 @desc: LLM 节点:单次调用大模型(不执行工具),输出 content/reasoning_content/refusal/is_refusal/usage/tool_calls/finish_reason
 """
 
+import asyncio
 import contextlib
 import json
 
@@ -61,6 +62,9 @@ def is_valid(self, *, raise_exception=False):
 
 def _get_default_model_params_setting(model_id):
     model = QuerySet(Model).filter(id=model_id).first()
+    if model is None:
+        # 模型已删除或不可见(如默认模型/引用变量指向了失效的模型)
+        raise AppApiException(500, _("The model of the node does not exist"))
     credential = get_model_credential(model.provider, model.model_type, model.model_name)
     return credential.get_model_params_setting_form(model.model_name).get_default_form_data()
 
@@ -290,8 +294,14 @@ async def _stream_response(self, model, chat_model, message_list):
                 )
             )
             # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接
-            self._write_final_context(
-                chat_model, message_list, full, reasoning_content, finish_reason, tool_call_accumulator.tool_calls()
+            await asyncio.to_thread(
+                self._write_final_context,
+                chat_model,
+                message_list,
+                full,
+                reasoning_content,
+                finish_reason,
+                tool_call_accumulator.tool_calls(),
             )
             with contextlib.suppress(Exception):
                 await response.aclose()

apps/application/workflow/nodes/question_node/question_node.py

@@ -5,6 +5,7 @@
 @desc:
 """
 
+import asyncio
 import re
 import contextlib
 from functools import reduce
@@ -173,7 +174,7 @@ async def _stream_response(self, response, chat_model, message_list, is_result):
                     )
         # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接
         finally:
-            self._write_final_context(chat_model, message_list, answer)
+            await asyncio.to_thread(self._write_final_context, chat_model, message_list, answer)
             with contextlib.suppress(Exception):
                 await response.aclose()
 

apps/application/workflow/nodes/video_understand_node/video_understand_node.py

@@ -1,4 +1,5 @@
 # coding=utf-8
+import asyncio
 import contextlib
 import uuid_utils.compat as uuid
 from functools import reduce
@@ -256,7 +257,9 @@ async def _stream_response(
 
         # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接
         finally:
-            self._write_final_context(chat_model, message_list, question, answer, reasoning_content)
+            await asyncio.to_thread(
+                self._write_final_context, chat_model, message_list, question, answer, reasoning_content
+            )
             with contextlib.suppress(Exception):
                 await response.aclose()