fix(agent): bind runtime tasks to app lifespan (#6303)

app/agent/__init__.py

@@ -2546,6 +2546,8 @@ class AgentManager:
             except asyncio.CancelledError:
                 pass
         self._session_workers.clear()
+        for queue in list(self._session_queues.values()):
+            self._discard_queued_messages(queue)
         self._session_queues.clear()
         self._session_last_used.clear()
         for agent in list(self.active_agents.values()):

app/startup/agent_initializer.py

@@ -1,6 +1,3 @@
-import asyncio
-import threading
-
 from app.agent import agent_manager
 from app.core.config import settings
 from app.log import logger
@@ -51,37 +48,16 @@ class AgentInitializer:
 agent_initializer = AgentInitializer()
 
 
-def init_agent():
+async def init_agent() -> bool:
     """
-    初始化AI智能体（同步版本，用于在后台线程中运行）
+    在应用事件循环中初始化AI智能体。
     """
     try:
         if not settings.AI_AGENT_ENABLE:
             logger.info("AI智能体功能未启用")
             return True
 
-        # 在新的事件循环中初始化AI智能体管理器
-        def run_init():
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            try:
-                success = loop.run_until_complete(agent_initializer.initialize())
-                if success:
-                    logger.info("AI智能体管理器初始化成功")
-                else:
-                    logger.error("AI智能体管理器初始化失败")
-                return success
-            except Exception as err:
-                logger.error(f"AI智能体管理器初始化失败: {err}")
-                return False
-            finally:
-                loop.close()
-
-        # 在后台线程中初始化
-        init_thread = threading.Thread(target=run_init, daemon=True)
-        init_thread.start()
-
-        return True
+        return await agent_initializer.initialize()
 
     except Exception as e:
         logger.error(f"初始化AI智能体时发生错误: {e}")

app/startup/lifecycle.py

@@ -80,7 +80,7 @@ async def lifespan(app: FastAPI):
     # 初始化路由
     init_routers(app)
     # 初始化模块
-    init_modules()
+    await init_modules()
     if settings.MOVIEPILOT_SAFE_MODE:
         print("MoviePilot safe mode enabled: skip plugins, scheduler, monitor, commands and workflow.")
     else:

app/startup/modules_initializer.py

@@ -154,7 +154,7 @@ async def stop_modules():
     await run_step("临时文件", clear_temp)
 
 
-def init_modules():
+async def init_modules():
     """
     启动模块
     """
@@ -178,7 +178,7 @@ def init_modules():
     MoviePilotServerHelper.get_user_uuid()
     MoviePilotServerHelper.get_github_user()
     # 初始化AI智能体
-    init_agent()
+    await init_agent()
     # 启动前端服务
     start_frontend()
     # 检查认证状态

tests/test_agent_lifecycle.py

@@ -0,0 +1,156 @@
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from app import agent as agent_module
+from app.agent import AgentManager
+from app.agent.memory import MemoryManager
+from app.startup import agent_initializer, modules_initializer
+
+
+@pytest.mark.anyio
+async def test_agent_entrypoint_initializes_on_calling_loop(monkeypatch) -> None:
+    """Agent 启动入口必须在应用主循环完成初始化。"""
+    current_loop = asyncio.get_running_loop()
+    initialized_loops = []
+    manager = AsyncMock()
+
+    async def initialize() -> None:
+        initialized_loops.append(asyncio.get_running_loop())
+
+    manager.initialize.side_effect = initialize
+    monkeypatch.setattr(agent_initializer.settings, "AI_AGENT_ENABLE", True)
+    monkeypatch.setattr(agent_initializer, "agent_manager", manager)
+    monkeypatch.setattr(
+        agent_initializer,
+        "agent_initializer",
+        agent_initializer.AgentInitializer(),
+    )
+
+    assert await agent_initializer.init_agent() is True
+    assert initialized_loops == [current_loop]
+    manager.initialize.assert_awaited_once_with()
+
+
+@pytest.mark.anyio
+async def test_agent_manager_background_tasks_share_owner_loop(monkeypatch) -> None:
+    """长期清理任务必须在同一循环创建、复用并完成关闭。"""
+    manager = AgentManager()
+    memory_manager = MemoryManager()
+    monkeypatch.setattr(agent_module, "memory_manager", memory_manager)
+    current_loop = asyncio.get_running_loop()
+
+    await manager.initialize()
+    idle_cleanup_task = manager._idle_cleanup_task
+    memory_cleanup_task = memory_manager.cleanup_task
+
+    assert idle_cleanup_task is not None
+    assert memory_cleanup_task is not None
+    assert idle_cleanup_task.get_loop() is current_loop
+    assert memory_cleanup_task.get_loop() is current_loop
+    assert not idle_cleanup_task.done()
+    assert not memory_cleanup_task.done()
+
+    await manager.initialize()
+    assert manager._idle_cleanup_task is idle_cleanup_task
+    assert memory_manager.cleanup_task is memory_cleanup_task
+
+    await manager.close()
+    await manager.close()
+    assert manager._idle_cleanup_task is None
+    assert memory_manager.cleanup_task is None
+    assert idle_cleanup_task.done()
+    assert memory_cleanup_task.done()
+
+
+@pytest.mark.anyio
+async def test_agent_entrypoint_reuses_tasks_and_closes_idempotently(
+        monkeypatch,
+) -> None:
+    """全局启停入口重复调用时必须复用任务并安全收口。"""
+    manager = AgentManager()
+    memory_manager = MemoryManager()
+    initializer = agent_initializer.AgentInitializer()
+    monkeypatch.setattr(agent_module, "memory_manager", memory_manager)
+    monkeypatch.setattr(agent_initializer.settings, "AI_AGENT_ENABLE", True)
+    monkeypatch.setattr(agent_initializer, "agent_manager", manager)
+    monkeypatch.setattr(agent_initializer, "agent_initializer", initializer)
+
+    assert await agent_initializer.init_agent() is True
+    idle_cleanup_task = manager._idle_cleanup_task
+    memory_cleanup_task = memory_manager.cleanup_task
+    assert await agent_initializer.init_agent() is True
+    assert manager._idle_cleanup_task is idle_cleanup_task
+    assert memory_manager.cleanup_task is memory_cleanup_task
+
+    await agent_initializer.stop_agent()
+    await agent_initializer.stop_agent()
+    assert initializer._initialized is False
+    assert manager._idle_cleanup_task is None
+    assert memory_manager.cleanup_task is None
+
+
+@pytest.mark.anyio
+async def test_agent_initialization_failure_does_not_stop_module_startup(
+        monkeypatch,
+) -> None:
+    """Agent 初始化异常只关闭该能力，基础模块仍继续完成启动。"""
+    manager = AsyncMock()
+    manager.initialize.side_effect = RuntimeError("agent init failed")
+    monkeypatch.setattr(agent_initializer.settings, "AI_AGENT_ENABLE", True)
+    monkeypatch.setattr(agent_initializer, "agent_manager", manager)
+    monkeypatch.setattr(
+        agent_initializer,
+        "agent_initializer",
+        agent_initializer.AgentInitializer(),
+    )
+    monkeypatch.setattr(modules_initializer, "init_agent", agent_initializer.init_agent)
+
+    for name in (
+        "DisplayHelper",
+        "DohHelper",
+        "SitesHelper",
+        "ResourceHelper",
+        "ModuleManager",
+    ):
+        monkeypatch.setattr(modules_initializer, name, MagicMock())
+    monkeypatch.setattr(modules_initializer, "user_auth", MagicMock())
+    monkeypatch.setattr(modules_initializer.EventManager, "start", MagicMock())
+    for name in (
+        "init_plugin_report",
+        "init_subscribe_report",
+        "get_user_uuid",
+        "get_github_user",
+    ):
+        monkeypatch.setattr(
+            modules_initializer.MoviePilotServerHelper,
+            name,
+            MagicMock(),
+        )
+    start_frontend = MagicMock()
+    check_auth = MagicMock()
+    monkeypatch.setattr(modules_initializer, "start_frontend", start_frontend)
+    monkeypatch.setattr(modules_initializer, "check_auth", check_auth)
+
+    await modules_initializer.init_modules()
+
+    manager.initialize.assert_awaited_once_with()
+    start_frontend.assert_called_once_with()
+    check_auth.assert_called_once_with()
+
+
+@pytest.mark.anyio
+async def test_disabled_agent_does_not_create_background_tasks(monkeypatch) -> None:
+    """Agent 未启用时启动入口不得创建运行时任务。"""
+    manager = AsyncMock()
+    monkeypatch.setattr(agent_initializer.settings, "AI_AGENT_ENABLE", False)
+    monkeypatch.setattr(agent_initializer, "agent_manager", manager)
+    monkeypatch.setattr(
+        agent_initializer,
+        "agent_initializer",
+        agent_initializer.AgentInitializer(),
+    )
+
+    assert await agent_initializer.init_agent() is True
+    manager.initialize.assert_not_awaited()

tests/test_agent_scheduled_tasks.py

@@ -961,6 +961,57 @@ async def test_agent_manager_records_cancelled_scheduled_task_as_failed() -> Non
     assert completed.run_count == 1
 
 
+@pytest.mark.anyio
+async def test_agent_manager_close_finishes_active_and_queued_scheduled_tasks() -> None:
+    """正常关闭必须取消同会话中正在执行和排队的持久任务。"""
+    user_id = f"shutdown-{uuid4().hex}"
+    session_id = f"session-{user_id}"
+    tasks = [
+        AgentTaskOper().add(
+            name=f"关闭中的后台检查 {index}",
+            content="检查资源并报告",
+            trigger_type="cron",
+            cron_expression="0 * * * *",
+            run_at=None,
+            user_id=user_id,
+            username="admin",
+            session_id=session_id,
+            channel=None,
+            source="api",
+            original_chat_id=None,
+        )
+        for index in range(2)
+    ]
+    manager = AgentManager()
+    started = asyncio.Event()
+
+    async def block_current_task(_task):
+        started.set()
+        await asyncio.Event().wait()
+
+    manager._process_message_internal = block_current_task
+    executions = [
+        asyncio.create_task(manager.execute_scheduled_task(task.id))
+        for task in tasks
+    ]
+    await asyncio.wait_for(started.wait(), timeout=1)
+    for _ in range(50):
+        if all(AgentTaskOper().get(task.id).last_status == "running" for task in tasks):
+            break
+        await asyncio.sleep(0)
+    assert all(AgentTaskOper().get(task.id).last_status == "running" for task in tasks)
+
+    await manager.close()
+    results = await asyncio.gather(*executions, return_exceptions=True)
+
+    assert all(isinstance(result, asyncio.CancelledError) for result in results)
+    for task in tasks:
+        completed = AgentTaskOper().get(task.id)
+        assert completed.last_status == "failed"
+        assert completed.last_result == "Agent 定时任务已取消"
+        assert completed.run_count == 1
+
+
 @pytest.mark.anyio
 async def test_cached_agent_clears_channel_for_background_task() -> None:
     """复用会话 Agent 时，后台任务必须覆盖上一轮保留的渠道信息。"""

tests/test_cache_system.py

@@ -2,6 +2,7 @@ import asyncio
 import os
 import threading
 import time
+from unittest.mock import AsyncMock
 
 from app.core.cache import (
     AsyncFileBackend,
@@ -149,13 +150,15 @@ def test_init_modules_does_not_clear_package_tool_cache(monkeypatch):
     monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "init_subscribe_report", lambda: None)
     monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_user_uuid", lambda: None)
     monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_github_user", lambda: None)
-    monkeypatch.setattr(modules_initializer, "init_agent", lambda: None)
+    init_agent = AsyncMock()
+    monkeypatch.setattr(modules_initializer, "init_agent", init_agent)
     monkeypatch.setattr(modules_initializer, "start_frontend", lambda: None)
     monkeypatch.setattr(modules_initializer, "check_auth", lambda: None)
 
-    modules_initializer.init_modules()
+    asyncio.run(modules_initializer.init_modules())
 
     assert called is False
+    init_agent.assert_awaited_once_with()
 
 def test_file_backend_delete_missing_key_is_noop(tmp_path):
     """

tests/test_lifecycle_shutdown.py

@@ -25,7 +25,6 @@ def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict:
 
     for name in (
         "init_routers",
-        "init_modules",
         "init_plugins",
         "init_scheduler",
         "init_monitor",
@@ -33,6 +32,7 @@ def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict:
         "init_workflow",
     ):
         monkeypatch.setattr(lifecycle, name, MagicMock())
+    monkeypatch.setattr(lifecycle, "init_modules", AsyncMock())
 
     system_chain = MagicMock()
     monkeypatch.setattr(lifecycle, "SystemChain", MagicMock(return_value=system_chain))
@@ -101,6 +101,7 @@ def test_lifespan_continues_after_each_shutdown_owner_failure(
     asyncio.run(run_lifespan())
 
     lifecycle.global_vars.stop_system.assert_called_once_with()
+    lifecycle.init_modules.assert_awaited_once_with()
     for step in shutdown_steps.values():
         _assert_completed_once(step)