From 1d2984b95fab46a952dbecc221dd164f031fe5d5 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sun, 23 Aug 2026 20:14:31 +0800 Subject: [PATCH] fix(runtime): enforce main event loop ownership (#6424) --- app/db/session.py | 4 +- app/runtime/config.py | 37 +++++------- app/startup/lifecycle/__init__.py | 13 ++-- tests/test_agent_image_capability.py | 6 +- tests/test_agent_image_support.py | 12 +++- tests/test_agent_interaction.py | 7 ++- tests/test_agent_message_routing.py | 35 ++++++++--- tests/test_global_event_loop.py | 60 +++++++++++++++++++ tests/test_lifecycle_shutdown.py | 5 ++ tests/test_plugin_monitor_lifecycle.py | 6 ++ tests/test_plugin_settlement_lifecycle.py | 6 ++ tests/test_telegram_typing_lifecycle.py | 6 +- tests/test_transfer_failed_retry_buttons.py | 14 +++-- ...ansfer_failure_notification_aggregation.py | 25 +++++--- 14 files changed, 181 insertions(+), 55 deletions(-) create mode 100644 tests/test_global_event_loop.py diff --git a/app/db/session.py b/app/db/session.py index f2791750b..e94911468 100644 --- a/app/db/session.py +++ b/app/db/session.py @@ -143,8 +143,8 @@ def _pooled_loop() -> Optional[Any]: 只认常驻主循环：它承载了绝大多数异步 DB 流量，且生命周期与进程一致， 池中连接不会因循环销毁而失效。 - 直接读 CURRENT_EVENT_LOOP 而不用 global_vars.loop——后者在未设置时会 - 新建一个事件循环，仅为判断就产生副作用是不可接受的。 + 直接读 CURRENT_EVENT_LOOP 而不用 global_vars.loop，避免主循环尚未就绪时 + 把正常的回退引擎选择转换成生命周期异常。 """ if not _async_pool_enabled(): return None diff --git a/app/runtime/config.py b/app/runtime/config.py index f9e6d0cbe..1cc0ec64c 100644 --- a/app/runtime/config.py +++ b/app/runtime/config.py @@ -1,4 +1,3 @@ -import asyncio import copy import json import os @@ -1359,18 +1358,8 @@ class GlobalVar(object): EMERGENCY_STOP_WORKFLOWS: List[int] = [] # 需应急停止文件整理 EMERGENCY_STOP_TRANSFER: List[str] = [] - # 当前事件循环 - CURRENT_EVENT_LOOP: AbstractEventLoop = None - - @classmethod - def _get_event_loop(cls) -> AbstractEventLoop: - """返回当前线程事件循环，缺失时创建并绑定新循环。""" - try: - return asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop + # 生命周期登记的主事件循环 + CURRENT_EVENT_LOOP: Optional[AbstractEventLoop] = None def stop_system(self): """ @@ -1461,19 +1450,21 @@ class GlobalVar(object): @property def loop(self) -> AbstractEventLoop: - """ - 当前循环 - """ - if self.CURRENT_EVENT_LOOP is None: - self.CURRENT_EVENT_LOOP = self._get_event_loop() - return self.CURRENT_EVENT_LOOP + """返回由应用生命周期登记的主事件循环。""" + loop = self.CURRENT_EVENT_LOOP + if loop is None or not loop.is_running() or loop.is_closed(): + raise RuntimeError("主事件循环尚未启动或已经停止") + return loop - def set_loop(self, loop: AbstractEventLoop): - """ - 设置循环 - """ + def set_loop(self, loop: AbstractEventLoop) -> None: + """登记承载主程序异步任务的事件循环。""" self.CURRENT_EVENT_LOOP = loop + def clear_loop(self, loop: AbstractEventLoop) -> None: + """仅在登记值仍为目标循环时清除主事件循环。""" + if self.CURRENT_EVENT_LOOP is loop: + self.CURRENT_EVENT_LOOP = None + # 全局标识 global_vars = GlobalVar() diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 493516afa..73ed3efef 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -368,14 +368,14 @@ async def lifespan(app: FastAPI): """ health = get_application_health(app) health.begin_startup() + main_loop = asyncio.get_running_loop() try: validate_process_topology( workers=settings.API_WORKERS, safe_mode=settings.MOVIEPILOT_SAFE_MODE, ) print("Starting up...") - # 存储当前循环 - global_vars.set_loop(asyncio.get_event_loop()) + global_vars.set_loop(main_loop) components = build_lifecycle_components(app) enabled_components = tuple( component @@ -411,6 +411,8 @@ async def lifespan(app: FastAPI): await stop_task_registry(app) except Exception as cleanup_error: logger.error(f"启动失败后的后台任务清理失败：{cleanup_error}") + finally: + global_vars.clear_loop(main_loop) raise try: # 在此处 yield，表示应用已经启动，控制权交回 FastAPI 主事件循环 @@ -435,5 +437,8 @@ async def lifespan(app: FastAPI): component.stop_timeout_seconds, ) finally: - # 日志最后关闭，确保其他组件的收尾信息已写入文件 - LoggerManager.shutdown() + try: + # 日志最后关闭，确保其他组件的收尾信息已写入文件 + LoggerManager.shutdown() + finally: + global_vars.clear_loop(main_loop) diff --git a/tests/test_agent_image_capability.py b/tests/test_agent_image_capability.py index bf6124f7c..3bafd17b1 100644 --- a/tests/test_agent_image_capability.py +++ b/tests/test_agent_image_capability.py @@ -1,5 +1,5 @@ from dataclasses import replace -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, Mock, patch import pytest @@ -7,7 +7,7 @@ from app.agent import MoviePilotAgent from app.agent.llm import AgentCapabilityManager, LLMHelper from app.agent.llm.provider import LLMProviderManager from app.chain.message import MessageChain -from app.runtime.config import settings +from app.runtime.config import global_vars, settings from app.schemas.types import NotificationChannel @@ -89,6 +89,8 @@ def test_handle_ai_message_routes_text_only_model_images_to_files( monkeypatch.setattr(settings, "LLM_SUPPORT_IMAGE_INPUT", True) monkeypatch.setattr(settings, "LLM_PROVIDER", "minimax") monkeypatch.setattr(settings, "LLM_MODEL", "MiniMax-M2.7") + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + monkeypatch.setattr(global_vars, "CURRENT_EVENT_LOOP", loop) # 测试绕过完整启动组合根，按需装配 llm_helper provider 以走真实能力判断 import app.application.agent as agent_facade diff --git a/tests/test_agent_image_support.py b/tests/test_agent_image_support.py index f4828d326..164713eab 100644 --- a/tests/test_agent_image_support.py +++ b/tests/test_agent_image_support.py @@ -16,7 +16,7 @@ from app.agent.tools.impl.send_local_file import SendLocalFileInput from app.agent import MoviePilotAgent, AgentChain from app.agent.llm import AgentCapabilityManager from app.chain.message import MessageChain -from app.runtime.config import settings +from app.runtime.config import global_vars, settings from app.agent.llm import LLMHelper from app.modules.discord import DiscordModule from app.modules.qqbot import QQBotModule @@ -454,7 +454,10 @@ class AgentImageSupportTest(unittest.TestCase): ai_agent_enable=True, ) - with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object( + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( + settings, "AI_AGENT_ENABLE", True + ), patch.object( settings, "LLM_SUPPORT_IMAGE_INPUT", False ), patch( "app.chain.message.supports_image_input", return_value=False @@ -508,7 +511,10 @@ class AgentImageSupportTest(unittest.TestCase): ai_agent_enable=True, ) - with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object( + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( + settings, "AI_AGENT_ENABLE", True + ), patch.object( chain, "_get_or_create_session_id", return_value="session-1" ), patch( "app.chain.message.get_running_agent_manager" diff --git a/tests/test_agent_interaction.py b/tests/test_agent_interaction.py index 9f404df23..04621b72d 100644 --- a/tests/test_agent_interaction.py +++ b/tests/test_agent_interaction.py @@ -17,7 +17,7 @@ from app.application.messaging.agent import ( ) from app.application.messaging.interaction import InteractionContext from app.chain.message import MessageChain -from app.runtime.config import settings +from app.runtime.config import global_vars, settings from app.schemas.types import NotificationChannel @@ -193,7 +193,10 @@ class TestAgentInteraction(unittest.TestCase): ], ) - with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object( + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( + settings, "AI_AGENT_ENABLE", True + ), patch.object( chain.messagehelper, "put" ) as message_put, patch.object( chain.messageoper, "add" diff --git a/tests/test_agent_message_routing.py b/tests/test_agent_message_routing.py index b28637acb..cd040dd70 100644 --- a/tests/test_agent_message_routing.py +++ b/tests/test_agent_message_routing.py @@ -11,7 +11,7 @@ from app.agent.tools.impl.ask_user_choice import ( ) from app.agent.tools.impl.send_message import SendMessageTool from app.chain.message import MessageChain -from app.runtime.config import settings +from app.runtime.config import global_vars, settings from app.db import SessionFactory from app.db.oper.message import MessageOper from app.db.models.message import Message @@ -28,6 +28,13 @@ def _clear_messages() -> None: db.commit() +def _running_loop_stub() -> Mock: + """提供满足主程序生命周期合同的事件循环替身。""" + return Mock( + **{"is_running.return_value": True, "is_closed.return_value": False} + ) + + def test_explicit_ai_message_bypasses_pending_media_interaction(): """显式 /ai 消息应绕过误触发的媒体交互状态并回到 Agent 会话。""" chain = MessageChain() @@ -69,7 +76,9 @@ def test_explicit_ai_message_is_not_recorded_to_message_history(): chain.runtime_config = replace(chain.runtime_config, ai_agent_enable=True) manager = Mock(process_message=AsyncMock()) - with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object( + with patch.object( + global_vars, "CURRENT_EVENT_LOOP", _running_loop_stub() + ), patch.object(settings, "AI_AGENT_ENABLE", True), patch.object( chain, "_record_user_message" ) as record_user_message, patch( "app.chain.message.get_running_agent_manager", return_value=manager @@ -101,7 +110,11 @@ def test_agent_queue_full_is_reported_to_the_originating_channel(): coro.close() return failed - with patch.object(settings, "AI_AGENT_ENABLE", True), patch( + with patch.object( + global_vars, "CURRENT_EVENT_LOOP", _running_loop_stub() + ), patch.object( + settings, "AI_AGENT_ENABLE", True + ), patch( "app.chain.message.get_running_agent_manager", return_value=manager ), patch( "app.chain.message.asyncio.run_coroutine_threadsafe", @@ -127,7 +140,9 @@ def test_message_chain_passes_stable_channel_admin_principal_to_agent(): chain.runtime_config = replace(chain.runtime_config, ai_agent_enable=True) manager = Mock(process_message=AsyncMock()) - with patch.object(settings, "AI_AGENT_ENABLE", True), patch( + with patch.object( + global_vars, "CURRENT_EVENT_LOOP", _running_loop_stub() + ), patch.object(settings, "AI_AGENT_ENABLE", True), patch( "app.chain.message.get_running_agent_manager", return_value=manager ), patch( "app.chain.message.asyncio.run_coroutine_threadsafe", @@ -151,7 +166,9 @@ def test_message_chain_does_not_trust_channel_display_username(): chain.runtime_config = replace(chain.runtime_config, ai_agent_enable=True) manager = Mock(process_message=AsyncMock()) - with patch.object(settings, "AI_AGENT_ENABLE", True), patch( + with patch.object( + global_vars, "CURRENT_EVENT_LOOP", _running_loop_stub() + ), patch.object(settings, "AI_AGENT_ENABLE", True), patch( "app.chain.message.get_running_agent_manager", return_value=manager ), patch( "app.chain.message.asyncio.run_coroutine_threadsafe", @@ -175,7 +192,9 @@ def test_message_chain_uses_same_admin_contract_for_slack(): chain.runtime_config = replace(chain.runtime_config, ai_agent_enable=True) manager = Mock(process_message=AsyncMock()) - with patch.object(settings, "AI_AGENT_ENABLE", True), patch( + with patch.object( + global_vars, "CURRENT_EVENT_LOOP", _running_loop_stub() + ), patch.object(settings, "AI_AGENT_ENABLE", True), patch( "app.chain.message.get_running_agent_manager", return_value=manager ), patch( "app.chain.message.asyncio.run_coroutine_threadsafe", @@ -295,7 +314,9 @@ def test_agent_choice_callback_is_not_recorded_to_message_history(): manager = Mock(process_message=AsyncMock()) try: - with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object( + with patch.object( + global_vars, "CURRENT_EVENT_LOOP", _running_loop_stub() + ), patch.object(settings, "AI_AGENT_ENABLE", True), patch.object( chain, "_record_user_message" ) as record_user_message, patch.object( chain, "edit_message", return_value=True diff --git a/tests/test_global_event_loop.py b/tests/test_global_event_loop.py new file mode 100644 index 000000000..208225535 --- /dev/null +++ b/tests/test_global_event_loop.py @@ -0,0 +1,60 @@ +import asyncio + +import pytest + +from app.runtime.config import GlobalVar + + +def test_global_loop_requires_lifecycle_owner() -> None: + """启动前读取主循环不得隐式创建一个无法执行任务的循环。""" + runtime = GlobalVar() + runtime.CURRENT_EVENT_LOOP = None + + with pytest.raises(RuntimeError, match="主事件循环尚未启动或已经停止"): + _ = runtime.loop + + assert runtime.CURRENT_EVENT_LOOP is None + + +def test_global_loop_rejects_closed_owner() -> None: + """已关闭的生命周期 owner 不得继续接收跨线程任务。""" + runtime = GlobalVar() + loop = asyncio.new_event_loop() + runtime.set_loop(loop) + loop.close() + + with pytest.raises(RuntimeError, match="主事件循环尚未启动或已经停止"): + _ = runtime.loop + + +def test_global_loop_rejects_owner_that_is_not_running() -> None: + """未运行的循环不得成为跨线程任务投递目标。""" + runtime = GlobalVar() + loop = asyncio.new_event_loop() + try: + runtime.set_loop(loop) + + with pytest.raises(RuntimeError, match="主事件循环尚未启动或已经停止"): + _ = runtime.loop + finally: + loop.close() + + +def test_clear_global_loop_preserves_new_owner() -> None: + """迟到的旧生命周期清理不得清除后来登记的循环。""" + runtime = GlobalVar() + previous = asyncio.new_event_loop() + + async def verify() -> None: + current = asyncio.get_running_loop() + runtime.set_loop(current) + runtime.clear_loop(previous) + assert runtime.loop is current + + runtime.clear_loop(current) + assert runtime.CURRENT_EVENT_LOOP is None + + try: + asyncio.run(verify()) + finally: + previous.close() diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index caeed2e16..87e7fe811 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -21,6 +21,7 @@ def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict: """隔离 lifespan 的外部依赖，并按名称注入一个关闭失败""" monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False) monkeypatch.setattr(lifecycle.global_vars, "set_loop", MagicMock()) + monkeypatch.setattr(lifecycle.global_vars, "clear_loop", MagicMock()) monkeypatch.setattr(lifecycle.global_vars, "stop_system", MagicMock()) for name in ( @@ -134,6 +135,8 @@ def test_lifespan_normal_mode_starts_full_runtime(monkeypatch): asyncio.run(run_lifespan()) + configured_loop = lifecycle.global_vars.set_loop.call_args.args[0] + lifecycle.global_vars.clear_loop.assert_called_once_with(configured_loop) lifecycle.init_modules.assert_awaited_once_with() lifecycle.prepare_database_component.assert_called_once() lifecycle.configure_plugin_services.assert_called_once_with() @@ -411,6 +414,8 @@ def test_lifespan_fails_fast_when_async_engine_cannot_be_built(monkeypatch): with pytest.raises(RuntimeError, match="no async driver"): asyncio.run(run_lifespan()) + configured_loop = lifecycle.global_vars.set_loop.call_args.args[0] + lifecycle.global_vars.clear_loop.assert_called_once_with(configured_loop) # 失败要发生在任何东西被初始化之前，否则模块起来了却没人关：关停块在 yield 处才开始 lifecycle.init_routers.assert_not_called() lifecycle.init_modules.assert_not_called() diff --git a/tests/test_plugin_monitor_lifecycle.py b/tests/test_plugin_monitor_lifecycle.py index 11a621ed1..eafe9db42 100644 --- a/tests/test_plugin_monitor_lifecycle.py +++ b/tests/test_plugin_monitor_lifecycle.py @@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest from app.foundation.singleton import Singleton +from app.runtime.config import global_vars from app.runtime.extensions.plugin.dependency import ( PluginDependencyClassification, PluginDependencyInstallResult, @@ -135,6 +136,11 @@ def _patch_sync_plugins(monkeypatch, manager: MagicMock) -> MagicMock: return task_func() register = MagicMock() + monkeypatch.setattr( + global_vars, + "CURRENT_EVENT_LOOP", + asyncio.get_running_loop(), + ) monkeypatch.setattr(plugins_initializer, "configure_plugin_services", lambda: None) monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager) monkeypatch.setattr(plugins_initializer, "execute_task", execute) diff --git a/tests/test_plugin_settlement_lifecycle.py b/tests/test_plugin_settlement_lifecycle.py index b84836789..28ffd86bd 100644 --- a/tests/test_plugin_settlement_lifecycle.py +++ b/tests/test_plugin_settlement_lifecycle.py @@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from app.runtime.config import global_vars from app.startup import lifecycle @@ -23,6 +24,11 @@ async def test_runtime_ready_waits_for_scheduler_and_command_refresh(monkeypatch return [] monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False) + monkeypatch.setattr( + global_vars, + "CURRENT_EVENT_LOOP", + asyncio.get_running_loop(), + ) monkeypatch.setattr(lifecycle, "get_plugin_manager", lambda: manager) monkeypatch.setattr(lifecycle, "sync_plugins", sync_plugins) monkeypatch.setattr(lifecycle, "execute_task", execute_task) diff --git a/tests/test_telegram_typing_lifecycle.py b/tests/test_telegram_typing_lifecycle.py index 31939f58d..83e0e30f0 100644 --- a/tests/test_telegram_typing_lifecycle.py +++ b/tests/test_telegram_typing_lifecycle.py @@ -12,6 +12,7 @@ from app.chain.message import MessageChain from app.command import Command, _finish_command_processing_status from app.modules.telegram import TelegramModule from app.modules.telegram.telegram import Telegram +from app.runtime.config import global_vars from app.schemas.types import NotificationChannel @@ -263,7 +264,10 @@ class TestTelegramTypingLifecycle(unittest.TestCase): ai_agent_enable=True, ) - with patch.object(chain, "_record_user_message"), patch.object( + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( + chain, "_record_user_message" + ), patch.object( chain, "_mark_message_processing_started" ) as start_status, patch( "app.chain.message.get_running_agent_manager", diff --git a/tests/test_transfer_failed_retry_buttons.py b/tests/test_transfer_failed_retry_buttons.py index 4d98edb3f..41a504421 100644 --- a/tests/test_transfer_failed_retry_buttons.py +++ b/tests/test_transfer_failed_retry_buttons.py @@ -4,7 +4,7 @@ import sys from dataclasses import replace from types import ModuleType from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import Mock, patch sys.modules.setdefault("qbittorrentapi", ModuleType("qbittorrentapi")) setattr(sys.modules["qbittorrentapi"], "TorrentFilesList", list) @@ -15,7 +15,7 @@ sys.modules.setdefault("psutil", ModuleType("psutil")) from app.chain.message import MessageChain from app.chain.transfer import TransferChain from app.application.messaging.interaction import InteractionContext -from app.runtime.config import settings +from app.runtime.config import global_vars, settings from app.schemas.types import NotificationChannel @@ -132,7 +132,10 @@ class TestTransferFailedRetryButtons(unittest.TestCase): """关闭被调度的协程：测试中事件循环未运行，不关闭会残留 never-awaited 警告。""" coro.close() - with patch.object(settings, "AI_AGENT_ENABLE", True): + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( + settings, "AI_AGENT_ENABLE", True + ): with patch( "app.chain._transfer.TransferHistoryOper" ) as history_oper_cls, patch( @@ -209,7 +212,10 @@ class TestTransferFailedRetryButtons(unittest.TestCase): from app.agent.prompt.transfer_redo import build_manual_redo_prompt manager = SimpleNamespace(run_background_prompt=fake_run_background_prompt) - with patch.object(settings, "AI_AGENT_ENABLE", True): + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( + settings, "AI_AGENT_ENABLE", True + ): with patch( "app.chain._transfer.TransferHistoryOper" ) as history_oper_cls, patch( diff --git a/tests/test_transfer_failure_notification_aggregation.py b/tests/test_transfer_failure_notification_aggregation.py index 868655da8..0e286f8e5 100644 --- a/tests/test_transfer_failure_notification_aggregation.py +++ b/tests/test_transfer_failure_notification_aggregation.py @@ -1,4 +1,4 @@ -from unittest.mock import Mock +from unittest.mock import Mock, patch from types import SimpleNamespace from app.chain import transfer as transfer_module @@ -48,6 +48,16 @@ class _Loop: self.timers.append(timer) return timer + @staticmethod + def is_running() -> bool: + """该替身代表由生命周期持有的运行中循环。""" + return True + + @staticmethod + def is_closed() -> bool: + """该替身在用例期间保持可用。""" + return False + def _task(*, episode: int, download_hash: str = "hash-1") -> TransferTask: """构造同一媒体不同剧集的整理任务。""" @@ -153,12 +163,13 @@ def test_enabled_queue_uses_shared_group_key(): message="整理失败", transfer_type="copy", ) - loop = transfer_module.global_vars.loop - chain.queue_failed_transfer_notification( - task=task, - transferinfo=transferinfo, - history_id=22, - ) + loop = _Loop() + with patch.object(transfer_module.global_vars, "CURRENT_EVENT_LOOP", loop): + chain.queue_failed_transfer_notification( + task=task, + transferinfo=transferinfo, + history_id=22, + ) chain.failure_notification_aggregator.schedule.assert_called_once() kwargs = chain.failure_notification_aggregator.schedule.call_args.kwargs