""" 验证 Agent 后台任务输出在前台会话和通知渠道中的处理方式。 """ from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from langchain_core.messages import AIMessage, HumanMessage from app.agent.contracts import ReplyMode from app.agent.manager import AgentManager from app.agent.memory import memory_manager from app.agent.middleware.invocation import GET_TOOL_EXECUTION_NAME, InvocationMiddleware from app.agent.middleware.memory import SEARCH_MEMORY_TOOL_NAME from app.agent.middleware.output import READ_TOOL_RESULT_NAME, ToolOutputMiddleware from app.agent.middleware.plan import PLAN_TOOL_NAME, PlanMiddleware from app.agent.middleware.selection import TOOL_DISCOVERY_NAME, ToolSelectorMiddleware from app.agent.middleware.skills import SKILL_TOOL_NAME from app.agent.middleware.subagents import ( SUBAGENT_CONTROL_TOOL_NAME, SUBAGENT_TASK_TOOL_NAME, ) from app.agent.orchestrator import ( HEARTBEAT_SESSION_PREFIX, UNSUPPORTED_IMAGE_INPUT_MESSAGE, MoviePilotAgent, ) from app.agent.session import _MessageTask from app.agent.tools.factory import MoviePilotToolFactory from app.foundation.identity import SYSTEM_INTERNAL_USER_ID from app.runtime.config import settings pytestmark = pytest.mark.anyio class _FakeGraphState: """ 保存测试所需的消息快照，避免访问真实持久化。 """ def __init__(self, messages): """ 保存该测试图的预设消息或故障。 """ self.values = {"messages": messages} class _FakeAgent: """ 提供可控的非流式图结果，不调用真实模型。 """ def __init__(self, messages): """ 保存该测试图的预设消息或故障。 """ self._messages = messages async def ainvoke(self, _payload, config=None): """ 返回预设结果或抛出故障，不执行外部调用。 """ return None def get_state(self, _config): """ 提供宿主结束处理所需的可恢复快照。 """ return _FakeGraphState(self._messages) class _FakeFailingAgent: """ 在调用边界注入故障，验证宿主错误回执。 """ def __init__(self, error): """ 保存该测试图的预设消息或故障。 """ self._error = error async def ainvoke(self, _payload, config=None): """ 返回预设结果或抛出故障，不执行外部调用。 """ raise self._error def get_state(self, _config): """ 提供宿主结束处理所需的可恢复快照。 """ return _FakeGraphState([]) class _FakeStreamingFailingAgent(_FakeFailingAgent): """ 在流式迭代边界注入同一故障。 """ async def astream(self, _messages, **_kwargs): """ 维持异步流接口，按测试预设结束或失败。 """ if _kwargs.get("__keep_async_generator__"): yield None raise self._error class _FakeStreamingAgent(_FakeAgent): """ 不产生额外 token，供测试读取最终图状态。 """ async def astream(self, _messages, **_kwargs): """ 维持异步流接口，按测试预设结束或失败。 """ if _kwargs.get("__keep_async_generator__"): yield None class StreamChunkTimeoutError(RuntimeError): """ 模拟 langchain_openai 的流式分块超时异常。 """ def _fake_skills_middleware(tool=None): """ 构造带 tools 属性的 SkillsMiddleware 测试替身。 """ return SimpleNamespace(name="skills", tools=[] if tool is None else [tool]) def _fake_memory_middleware(tool=None): """ 构造带 tools 属性的统一 MemoryMiddleware 测试替身。 """ return SimpleNamespace(name="memory", tools=[] if tool is None else [tool]) def _capture_tool_selector(captured, **kwargs): """ 保留真实发现工具与筛选合同，只替换无需执行的测试模型实例。 """ selector_options = {**kwargs, "model": None} selector = ToolSelectorMiddleware(**selector_options) selector.model = kwargs["model"] captured.update( selector=selector, selection_tools=selector.selection_tools, always_include=selector.always_include, enable_discovery=kwargs["enable_discovery"], ) return selector def _assert_internal_tool_registration(created, captured): """ 核实内部计划、续读和发现工具的目录身份、常驻筛选及外层顺序。 """ middlewares = created["middleware"] plan = next(item for item in middlewares if isinstance(item, PlanMiddleware)) output = next(item for item in middlewares if isinstance(item, ToolOutputMiddleware)) selector = captured["selector"] policy = middlewares[0] assert policy.name == "AgentPolicyMiddleware" assert middlewares[1].name == "CodeExecutionMiddleware" assert middlewares[2] is output assert not any(isinstance(item, InvocationMiddleware) for item in middlewares) assert policy.catalog.resolve_unique(GET_TOOL_EXECUTION_NAME) is None assert captured["enable_discovery"] is True assert {PLAN_TOOL_NAME, READ_TOOL_RESULT_NAME, TOOL_DISCOVERY_NAME} <= set(selector.always_include) for tool in [*plan.tools, *output.tools, *selector.tools]: assert policy.catalog.resolve_unique(tool.name).tool is tool assert tool in selector.selection_tools assert middlewares.index(plan) < middlewares.index(selector) assert middlewares.index(selector) == len(middlewares) - 5 assert middlewares[-4].name == "CodeCaptureMiddleware" assert middlewares[-3].name == "FinalRequestCompactionMiddleware" assert middlewares[-2].name == "VisionMiddleware" assert middlewares[-1] == "usage" class TestAgentBackgroundOutput: """ 验证后台任务的回复策略、流式结束行为和工具装配。 """ async def test_non_streaming_image_unsupported_error_sends_friendly_notice(self): """ 非流式图片能力错误只向用户交付友好说明。 """ agent = MoviePilotAgent(session_id="image-test", user_id="user-1") agent.channel = "Telegram" agent.source = "telegram-test" agent._tool_context = {"user_reply_sent": False} agent._streamed_output = "" agent.stream_handler = SimpleNamespace( stop_streaming=AsyncMock(return_value=(False, "")) ) agent._should_stream = lambda: False agent._create_agent = AsyncMock( return_value=_FakeFailingAgent( RuntimeError("No endpoints found that support image input") ) ) agent.send_agent_message = AsyncMock() result, _ = await agent._execute_agent( [ HumanMessage( content=[ {"type": "text", "text": "看看这张图"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64,xxx"}}, ] ) ] ) assert result == UNSUPPORTED_IMAGE_INPUT_MESSAGE agent.send_agent_message.assert_awaited_once_with( UNSUPPORTED_IMAGE_INPUT_MESSAGE, title="" ) assert agent._streamed_output == UNSUPPORTED_IMAGE_INPUT_MESSAGE async def test_streaming_image_unsupported_error_sends_friendly_notice(self): """ 流式图片能力错误沿用同一用户可见说明。 """ agent = MoviePilotAgent(session_id="image-test", user_id="user-1") agent.channel = "Telegram" agent.source = "telegram-test" agent._tool_context = {"user_reply_sent": False} agent._streamed_output = "" agent.stream_handler = SimpleNamespace( set_dispatch_policy=lambda allow_dispatch_without_context=False: None, start_streaming=AsyncMock(), flush_pending_tool_summary=lambda: "", stop_streaming=AsyncMock(return_value=(False, "")), ) agent._should_stream = lambda: True agent._create_agent = AsyncMock( return_value=_FakeStreamingFailingAgent( RuntimeError("Error code: 404 - No endpoints found that support image input") ) ) agent.send_agent_message = AsyncMock() result, _ = await agent._execute_agent( [ HumanMessage( content=[ {"type": "text", "text": "看看这张图"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64,xxx"}}, ] ) ] ) assert result == UNSUPPORTED_IMAGE_INPUT_MESSAGE agent.send_agent_message.assert_awaited_once_with( UNSUPPORTED_IMAGE_INPUT_MESSAGE, title="" ) async def test_streaming_model_chunk_timeout_sends_friendly_notice(self): """ 流式模型分块超时时应只把主错误信息发给用户。 """ agent = MoviePilotAgent(session_id="timeout-test", user_id="user-1") agent.channel = "Telegram" agent.source = "telegram-test" agent._tool_context = {"user_reply_sent": False} agent._streamed_output = "" agent.stream_handler = SimpleNamespace( set_dispatch_policy=lambda allow_dispatch_without_context=False: None, start_streaming=AsyncMock(), flush_pending_tool_summary=lambda: "", stop_streaming=AsyncMock(return_value=(False, "")), ) agent._should_stream = lambda: True raw_error = StreamChunkTimeoutError( "No streaming chunk received for 120.0s " "(model=mimo-v2.5-pro, chunks_received=1). " "Tune or disable via the `stream_chunk_timeout` constructor kwarg." ) agent._create_agent = AsyncMock( return_value=_FakeStreamingFailingAgent(raw_error) ) agent.send_agent_message = AsyncMock() result, _ = await agent._execute_agent([HumanMessage(content="测试超时")]) expected = "智能助手执行失败，请稍后重试" assert result == expected agent.send_agent_message.assert_awaited_once_with(expected, title="") sent_message = agent.send_agent_message.await_args.args[0] assert "No streaming chunk received for 120.0s" not in sent_message assert agent._streamed_output == expected async def test_streaming_success_stops_streaming_once(self): """ 流式正常完成时不应在 finally 中重复停止流式输出。 """ agent = MoviePilotAgent(session_id="stream-ok", user_id="user-1") agent.channel = "Telegram" agent.source = "telegram-test" agent._tool_context = {"user_reply_sent": False} agent._streamed_output = "" agent.stream_handler = SimpleNamespace( set_dispatch_policy=lambda allow_dispatch_without_context=False: None, start_streaming=AsyncMock(), flush_pending_tool_summary=lambda: "", stop_streaming=AsyncMock(return_value=(True, "已发送")), ) agent._should_stream = lambda: True agent._create_agent = AsyncMock( return_value=_FakeStreamingAgent([AIMessage(content="已发送")]) ) agent.send_agent_message = AsyncMock() await agent._execute_agent([HumanMessage(content="测试")]) agent.stream_handler.stop_streaming.assert_awaited_once() async def test_tool_sent_reply_persists_raw_agent_messages(self): """ 工具已发送用户回复时仍应保存可恢复的 Agent 原始消息。 """ agent = MoviePilotAgent(session_id="tool-reply", user_id="user-1") agent.channel = "Telegram" agent.source = "telegram-test" agent._tool_context = {"user_reply_sent": True} agent._streamed_output = "" agent.stream_handler = SimpleNamespace( stop_streaming=AsyncMock(return_value=(False, "")) ) agent._should_stream = lambda: False agent._create_agent = AsyncMock( return_value=_FakeAgent([AIMessage(content="消息已发送")]) ) agent.send_agent_message = AsyncMock() with patch.object( memory_manager, "async_save_agent_messages", new=AsyncMock(), ) as save_messages: await agent._execute_agent([HumanMessage(content="测试")]) save_messages.assert_called_once() _, kwargs = save_messages.call_args assert kwargs["session_id"] == "tool-reply" assert kwargs["user_id"] == "user-1" assert kwargs["messages"][0].content == "消息已发送" async def test_process_does_not_mutate_cached_agent_messages(self): """ 处理新消息时不应直接修改记忆缓存中的历史消息列表。 """ agent = MoviePilotAgent( session_id="cached-memory", user_id="user-1", channel="Telegram", source="telegram-test", ) cached_messages = [HumanMessage(content="上一轮")] captured = {} async def _execute_agent(messages): """ 记录模拟图实际收到的当前用户输入。 """ captured["messages"] = messages return "消息已发送", {} agent._execute_agent = AsyncMock(side_effect=_execute_agent) with ( patch.object( memory_manager, "async_get_agent_messages", new=AsyncMock(return_value=cached_messages), ), patch.object(agent, "prepare_chat_title", new=AsyncMock()), patch.object(agent, "_save_display_history_messages"), ): result = await agent.process("继续") assert result == "消息已发送" assert len(cached_messages) == 1 assert cached_messages is not captured["messages"] assert len(captured["messages"]) == 2 async def test_background_non_streaming_sends_when_reply_mode_dispatch(self): """ 显式派发模式允许后台任务发送最终回复。 """ agent = MoviePilotAgent(session_id="bg-test", user_id="system") agent.channel = None agent.source = None agent.reply_mode = ReplyMode.DISPATCH agent._tool_context = {"user_reply_sent": False} agent._streamed_output = "" agent.stream_handler = SimpleNamespace( stop_streaming=AsyncMock(return_value=(False, "")) ) agent._should_stream = lambda: False agent._create_agent = AsyncMock( return_value=_FakeAgent([AIMessage(content="后台结果")]) ) agent.send_agent_message = AsyncMock() with patch.object( memory_manager, "async_save_agent_messages", new=AsyncMock(), ) as save_messages: await agent._execute_agent([]) agent.send_agent_message.assert_awaited_once_with( "后台结果", title="MoviePilot助手" ) save_messages.assert_not_called() assert agent._streamed_output == "后台结果" async def test_background_non_streaming_captures_without_sending_when_capture_only(self): """ 显式捕获模式保留最终结果，但不发送消息或保存渠道历史。 """ agent = MoviePilotAgent(session_id="bg-test", user_id="system") agent.channel = None agent.source = None agent.reply_mode = ReplyMode.CAPTURE_ONLY agent._tool_context = {"user_reply_sent": False} agent._streamed_output = "" agent.stream_handler = SimpleNamespace( stop_streaming=AsyncMock(return_value=(False, "")) ) agent._should_stream = lambda: False agent._create_agent = AsyncMock( return_value=_FakeAgent([AIMessage(content="后台结果")]) ) agent.send_agent_message = AsyncMock() with patch.object(memory_manager, "save_agent_messages") as save_messages: await agent._execute_agent([]) agent.send_agent_message.assert_not_awaited() save_messages.assert_not_called() assert agent._streamed_output == "后台结果" async def test_heartbeat_check_jobs_captures_final_reply_and_keeps_message_tools(self): """ 心跳捕获最终回复但保留任务实际需要的消息工具。 """ manager = AgentManager() with ( patch("app.agent.tasks.load_jobs_metadata", new=AsyncMock(return_value=[{ "id": "job-1", "name": "测试任务", "description": "desc", "path": "/tmp/job-1/JOB.md", "schedule": "once", "status": "pending", "last_run": None, }])), patch.object(manager, "_build_heartbeat_prompt", return_value="HEARTBEAT"), patch.object(manager, "process_message", new=AsyncMock()) as process_message, ): await manager.heartbeat_check_jobs() process_message.assert_awaited_once() kwargs = process_message.await_args.kwargs assert kwargs["reply_mode"] == ReplyMode.CAPTURE_ONLY assert kwargs["allow_message_tools"] is True async def test_heartbeat_check_jobs_skips_when_no_active_jobs(self): """ 没有启用任务时不调用模型执行心跳。 """ manager = AgentManager() with ( patch("app.agent.tasks.load_jobs_metadata", new=AsyncMock(return_value=[])), patch.object(manager, "process_message", new=AsyncMock()) as process_message, ): await manager.heartbeat_check_jobs() process_message.assert_not_awaited() async def test_agent_manager_preserves_voice_input_flag(self): """ 会话队列执行时应把语音输入标记继续传给 Agent。 """ manager = AgentManager() agent = MoviePilotAgent(session_id="session-1", user_id="user-1") manager.active_agents["session-1"] = agent agent.process = AsyncMock(return_value="ok") task = _MessageTask( session_id="session-1", user_id="user-1", message="帮我推荐一部电影", has_audio_input=True, ) await manager._process_message_internal(task) agent.process.assert_awaited_once_with( "帮我推荐一部电影", images=None, files=None, has_audio_input=True, ) async def test_create_agent_disables_activity_recording_for_heartbeat_session(self): """ 心跳任务保留统一记忆能力，但不启用渠道活动记录。 """ agent = MoviePilotAgent( session_id=f"{HEARTBEAT_SESSION_PREFIX}test__", user_id="system", ) agent._initialize_tools = lambda: [] agent._initialize_subagent_tools = lambda: [] with ( patch.object(settings, "LLM_MAX_TOOLS", 0), patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())), patch("app.agent.orchestrator.prompt_manager.get_agent_prompt", return_value="PROMPT"), patch("app.agent.orchestrator.create_subagent_middlewares", return_value=([], [])), patch( "app.agent.tools.factory.MoviePilotToolFactory.get_tool_selector_always_include_names", return_value=[], ), patch( "app.agent.orchestrator.SkillsMiddleware", side_effect=lambda *args, **kwargs: _fake_skills_middleware(), ), patch("app.agent.orchestrator.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"), patch("app.agent.orchestrator.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime"), patch("app.agent.orchestrator.MemoryMiddleware", side_effect=lambda *args, **kwargs: "memory"), patch("app.agent.orchestrator.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary"), patch("app.agent.orchestrator.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch"), patch("app.agent.orchestrator.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"), patch("app.agent.orchestrator.InMemorySaver", return_value="checkpointer"), patch("app.agent.orchestrator.create_agent", side_effect=lambda **kwargs: kwargs), ): created = await agent._create_agent(streaming=False) assert [getattr(item, "name", item) for item in created["middleware"]] == [ "AgentPolicyMiddleware", "CodeExecutionMiddleware", "ToolOutputMiddleware", "skills", "jobs", "runtime", "PlanMiddleware", "memory", "ToolGuardrailsMiddleware", "patch", "CodeCaptureMiddleware", "FinalRequestCompactionMiddleware", "VisionMiddleware", "usage", ] async def test_create_agent_registers_skill_tool_from_middleware(self): """ SkillsMiddleware 暴露的 read_skill 应进入 Agent 工具和筛选候选。 """ captured = {} skill_tool = SimpleNamespace(name=SKILL_TOOL_NAME) agent = MoviePilotAgent(session_id="normal-session", user_id="system") agent._initialize_tools = lambda: [] agent._initialize_subagent_tools = lambda: [] def _tool_selector(**kwargs): """ 记录技能能力进入真实筛选目录后的参数。 """ return _capture_tool_selector(captured, **kwargs) with ( patch.object(settings, "LLM_MAX_TOOLS", 5), patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())), patch("app.agent.orchestrator.prompt_manager.get_agent_prompt", return_value="PROMPT"), patch("app.agent.orchestrator.create_subagent_middlewares", return_value=([], [])), patch( "app.agent.tools.factory.MoviePilotToolFactory.get_tool_selector_always_include_names", return_value=[], ), patch( "app.agent.orchestrator.SkillsMiddleware", side_effect=lambda *args, **kwargs: _fake_skills_middleware(skill_tool), ), patch("app.agent.orchestrator.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"), patch( "app.agent.orchestrator.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime", ), patch("app.agent.orchestrator.MemoryMiddleware", side_effect=lambda *args, **kwargs: "memory"), patch( "app.agent.orchestrator.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary", ), patch( "app.agent.orchestrator.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch", ), patch("app.agent.orchestrator.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"), patch("app.agent.orchestrator.ToolSelectorMiddleware", side_effect=_tool_selector), patch("app.agent.orchestrator.InMemorySaver", return_value="checkpointer"), patch("app.agent.orchestrator.create_agent", side_effect=lambda **kwargs: kwargs), ): created = await agent._create_agent(streaming=False) assert skill_tool in created["tools"] assert skill_tool in captured["selection_tools"] assert SKILL_TOOL_NAME in captured["always_include"] _assert_internal_tool_registration(created, captured) async def test_create_agent_disables_activity_recording_without_message_context(self): """ 无渠道信息的后台捕获任务不应启用活动记录。 """ agent = MoviePilotAgent( session_id="background-capture-session", user_id="system", output_callback=lambda _text: None, ) agent._initialize_tools = lambda: [] agent._initialize_subagent_tools = lambda: [] with ( patch.object(settings, "LLM_MAX_TOOLS", 0), patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())), patch("app.agent.orchestrator.prompt_manager.get_agent_prompt", return_value="PROMPT"), patch("app.agent.orchestrator.create_subagent_middlewares", return_value=([], [])), patch( "app.agent.tools.factory.MoviePilotToolFactory.get_tool_selector_always_include_names", return_value=[], ), patch( "app.agent.orchestrator.SkillsMiddleware", side_effect=lambda *args, **kwargs: _fake_skills_middleware(), ), patch("app.agent.orchestrator.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"), patch( "app.agent.orchestrator.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime", ), patch("app.agent.orchestrator.MemoryMiddleware", side_effect=lambda *args, **kwargs: "memory"), patch( "app.agent.orchestrator.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary", ), patch( "app.agent.orchestrator.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch", ), patch("app.agent.orchestrator.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"), patch("app.agent.orchestrator.InMemorySaver", return_value="checkpointer"), patch("app.agent.orchestrator.create_agent", side_effect=lambda **kwargs: kwargs), ): created = await agent._create_agent(streaming=False) assert [getattr(item, "name", item) for item in created["middleware"]] == [ "AgentPolicyMiddleware", "CodeExecutionMiddleware", "ToolOutputMiddleware", "skills", "jobs", "runtime", "PlanMiddleware", "memory", "ToolGuardrailsMiddleware", "patch", "CodeCaptureMiddleware", "FinalRequestCompactionMiddleware", "VisionMiddleware", "usage", ] def test_message_tool_is_not_always_included_by_tool_selector(self): """ 消息发送工具不应绕过工具筛选。 """ send_message_tool = SimpleNamespace(name="send_message") always_include = MoviePilotToolFactory.get_tool_selector_always_include_names( [send_message_tool] ) assert "send_message" not in always_include def test_memory_tool_is_always_included_by_tool_selector(self): """ 记忆检索工具应作为常驻候选，但不由全局工具工厂重复注册。 """ memory_tool = SimpleNamespace(name=SEARCH_MEMORY_TOOL_NAME) always_include = MoviePilotToolFactory.get_tool_selector_always_include_names( [memory_tool] ) assert SEARCH_MEMORY_TOOL_NAME in always_include async def test_create_agent_registers_memory_tool_from_middleware(self): """ MemoryMiddleware 暴露的工具应进入 Agent 工具和筛选候选。 """ captured = {} memory_tool = SimpleNamespace(name=SEARCH_MEMORY_TOOL_NAME) agent = MoviePilotAgent( session_id="normal-session", user_id="system", channel="Web", source="openai", ) agent._initialize_tools = lambda: [] agent._initialize_subagent_tools = lambda: [] def _tool_selector(**kwargs): """ 记录活动日志能力进入真实筛选目录后的参数。 """ return _capture_tool_selector(captured, **kwargs) with ( patch.object(settings, "LLM_MAX_TOOLS", 5), patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())), patch("app.agent.orchestrator.prompt_manager.get_agent_prompt", return_value="PROMPT"), patch("app.agent.orchestrator.create_subagent_middlewares", return_value=([], [])), patch( "app.agent.tools.factory.MoviePilotToolFactory.get_tool_selector_always_include_names", return_value=[], ), patch( "app.agent.orchestrator.SkillsMiddleware", side_effect=lambda *args, **kwargs: _fake_skills_middleware(), ), patch("app.agent.orchestrator.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"), patch( "app.agent.orchestrator.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime", ), patch( "app.agent.orchestrator.MemoryMiddleware", side_effect=lambda *args, **kwargs: _fake_memory_middleware(memory_tool), ), patch( "app.agent.orchestrator.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary", ), patch( "app.agent.orchestrator.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch", ), patch("app.agent.orchestrator.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"), patch("app.agent.orchestrator.ToolSelectorMiddleware", side_effect=_tool_selector), patch("app.agent.orchestrator.InMemorySaver", return_value="checkpointer"), patch("app.agent.orchestrator.create_agent", side_effect=lambda **kwargs: kwargs), ): created = await agent._create_agent(streaming=False) assert memory_tool in created["tools"] assert memory_tool in captured["selection_tools"] assert SEARCH_MEMORY_TOOL_NAME in captured["always_include"] _assert_internal_tool_registration(created, captured) async def test_create_agent_always_includes_subagent_tools(self): """ 工具筛选开启时应保留同步和异步子代理入口。 """ captured = {} agent = MoviePilotAgent(session_id="normal-session", user_id="system") agent._initialize_tools = lambda: [] agent._initialize_subagent_tools = lambda: [] def _tool_selector(**kwargs): """ 记录子代理能力进入真实筛选目录后的参数。 """ return _capture_tool_selector(captured, **kwargs) with ( patch.object(settings, "LLM_MAX_TOOLS", 5), patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())), patch("app.agent.orchestrator.prompt_manager.get_agent_prompt", return_value="PROMPT"), patch( "app.agent.orchestrator.create_subagent_middlewares", return_value=( ["subagent"], [ SimpleNamespace(name=SUBAGENT_TASK_TOOL_NAME), SimpleNamespace(name=SUBAGENT_CONTROL_TOOL_NAME), ], ), ), patch( "app.agent.tools.factory.MoviePilotToolFactory.get_tool_selector_always_include_names", return_value=[], ), patch( "app.agent.orchestrator.SkillsMiddleware", side_effect=lambda *args, **kwargs: _fake_skills_middleware(), ), patch("app.agent.orchestrator.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"), patch( "app.agent.orchestrator.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime", ), patch("app.agent.orchestrator.MemoryMiddleware", side_effect=lambda *args, **kwargs: "memory"), patch( "app.agent.orchestrator.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary", ), patch( "app.agent.orchestrator.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch", ), patch("app.agent.orchestrator.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"), patch("app.agent.orchestrator.ToolSelectorMiddleware", side_effect=_tool_selector), patch("app.agent.orchestrator.InMemorySaver", return_value="checkpointer"), patch("app.agent.orchestrator.create_agent", side_effect=lambda **kwargs: kwargs), ): created = await agent._create_agent(streaming=False) assert SUBAGENT_TASK_TOOL_NAME in captured["always_include"] assert SUBAGENT_CONTROL_TOOL_NAME in captured["always_include"] _assert_internal_tool_registration(created, captured) async def test_create_agent_uses_one_memory_middleware_for_normal_session(self): """ 普通渠道会话用同一个中间件提供记忆检索和活动记录。 """ agent = MoviePilotAgent( session_id="normal-session", user_id="system", channel="Web", source="openai", ) agent._initialize_tools = lambda: [] agent._initialize_subagent_tools = lambda: [] with ( patch.object(settings, "LLM_MAX_TOOLS", 0), patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())), patch("app.agent.orchestrator.prompt_manager.get_agent_prompt", return_value="PROMPT"), patch("app.agent.orchestrator.create_subagent_middlewares", return_value=([], [])), patch( "app.agent.tools.factory.MoviePilotToolFactory.get_tool_selector_always_include_names", return_value=[], ), patch( "app.agent.orchestrator.SkillsMiddleware", side_effect=lambda *args, **kwargs: _fake_skills_middleware(), ), patch("app.agent.orchestrator.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"), patch( "app.agent.orchestrator.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime", ), patch("app.agent.orchestrator.MemoryMiddleware", side_effect=lambda *args, **kwargs: "memory"), patch( "app.agent.orchestrator.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary", ), patch( "app.agent.orchestrator.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch", ), patch("app.agent.orchestrator.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"), patch("app.agent.orchestrator.InMemorySaver", return_value="checkpointer"), patch("app.agent.orchestrator.create_agent", side_effect=lambda **kwargs: kwargs), ): created = await agent._create_agent(streaming=False) assert [getattr(item, "name", item) for item in created["middleware"]] == [ "AgentPolicyMiddleware", "CodeExecutionMiddleware", "ToolOutputMiddleware", "skills", "jobs", "runtime", "PlanMiddleware", "memory", "ToolGuardrailsMiddleware", "patch", "CodeCaptureMiddleware", "FinalRequestCompactionMiddleware", "VisionMiddleware", "usage", ] async def test_run_background_prompt_forces_disable_message_tools_when_capture_only(self): """ 仅捕获的后台入口强制移除主动消息发送能力。 """ captured = {} manager = AgentManager() async def fake_process(task): """ 捕获后台参数而不触发真实 Agent 推理。 """ captured["message"] = task.message captured["reply_mode"] = task.reply_mode captured["allow_message_tools"] = task.allow_message_tools captured["user_id"] = task.user_id manager._process_message_internal = fake_process await manager.initialize() try: await manager.run_background_prompt( message="background task", reply_mode=ReplyMode.CAPTURE_ONLY, allow_message_tools=True, ) finally: await manager.close() assert captured["message"] == "background task" assert captured["reply_mode"] == ReplyMode.CAPTURE_ONLY assert captured["allow_message_tools"] is False assert captured["user_id"] == SYSTEM_INTERNAL_USER_ID