diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 95df59048..d74a0bc49 100644 --- a/app/agent/__init__.py +++ b/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()): diff --git a/app/startup/agent_initializer.py b/app/startup/agent_initializer.py index 21c9e7538..d64af28e6 100644 --- a/app/startup/agent_initializer.py +++ b/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}") diff --git a/app/startup/lifecycle.py b/app/startup/lifecycle.py index b0c1dade3..f4b57f338 100644 --- a/app/startup/lifecycle.py +++ b/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: diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index e8a66f54f..035213e56 100644 --- a/app/startup/modules_initializer.py +++ b/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() # 检查认证状态 diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py new file mode 100644 index 000000000..a9a108bfe --- /dev/null +++ b/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() diff --git a/tests/test_agent_scheduled_tasks.py b/tests/test_agent_scheduled_tasks.py index 033191364..6478a2109 100644 --- a/tests/test_agent_scheduled_tasks.py +++ b/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 时,后台任务必须覆盖上一轮保留的渠道信息。""" diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 4fb839703..e2467df47 100644 --- a/tests/test_cache_system.py +++ b/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): """ diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 25f88bc13..81690182c 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/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)