From d666134ed2a5de6c4c83713ac62e7ac0a4c72ad4 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 20 Jul 2026 08:54:32 +0800 Subject: [PATCH] Route autonomous agent tasks through notification broadcasting --- app/agent/__init__.py | 76 ++-- app/agent/middleware/jobs.py | 7 +- app/agent/prompt/System Core Prompt.txt | 2 +- app/agent/tools/base.py | 17 +- app/agent/tools/factory.py | 2 + app/agent/tools/impl/create_agent_task.py | 4 +- app/agent/tools/impl/delete_agent_task.py | 2 +- app/agent/tools/impl/query_agent_tasks.py | 8 +- app/agent/tools/impl/query_schedulers.py | 68 ++-- app/agent/tools/impl/run_agent_task.py | 78 ++++ app/agent/tools/impl/run_scheduler.py | 27 +- app/agent/tools/impl/update_agent_task.py | 2 +- app/agent/tools/tags.py | 1 + app/api/endpoints/agent.py | 4 +- app/scheduler.py | 15 + docs/mcp-api.md | 9 +- skills/moviepilot-cli/SKILL.md | 19 +- tests/test_agent_scheduled_tasks.py | 434 +++++++++++++++++++++- tests/test_builtin_skill_boundaries.py | 2 +- tests/test_web_agent_stream.py | 15 + 20 files changed, 676 insertions(+), 116 deletions(-) create mode 100644 app/agent/tools/impl/run_agent_task.py diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 7d834ee8..b6ec4ebe 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -1628,20 +1628,21 @@ class MoviePilotAgent: if not streaming_stopped: await self.stream_handler.stop_streaming() - async def send_agent_message(self, message: str, title: str = ""): + async def send_agent_message(self, message: str, title: str = "") -> None: """ - 通过原渠道发送消息给用户 + 发送 Agent 消息；后台任务不绑定原渠道，交由通知链广播。 """ + broadcast = self.is_background self._save_assistant_display_message_once(message) await AgentChain().async_post_message( Notification( - channel=self.channel, - source=self.source, + channel=None if broadcast else self.channel, + source=None if broadcast else self.source, mtype=NotificationType.Agent, - userid=self.user_id, - username=self.username, - original_message_id=self.original_message_id, - original_chat_id=self.original_chat_id, + userid=None if broadcast else self.user_id, + username=self.username or (settings.SUPERUSER if broadcast else None), + original_message_id=None if broadcast else self.original_message_id, + original_chat_id=None if broadcast else self.original_chat_id, title=title, text=message, save_history=False, @@ -2001,12 +2002,11 @@ class AgentManager: else: agent = self.active_agents[session_id] agent.user_id = task.user_id - if task.channel: - agent.channel = task.channel - if task.source: - agent.source = task.source - if task.username: - agent.username = task.username + # 每条队列任务都携带完整消息上下文，None 也必须覆盖，避免后台任务 + # 复用会话 Agent 时继续沿用上一条入站消息的渠道。 + agent.channel = task.channel + agent.source = task.source + agent.username = task.username agent.original_message_id = task.original_message_id agent.original_chat_id = task.original_chat_id agent.reply_mode = task.reply_mode @@ -2147,24 +2147,20 @@ class AgentManager: f"任务内容：{task.content}\n\n" "完成后请直接向用户报告本次执行结果；如果无法完成，请说明原因。" ) - has_message_context = bool(task.channel and task.source) success = True result = "" + notification_username = task.username or settings.SUPERUSER try: result = await self.process_message( session_id=task.session_id, user_id=task.user_id, message=task_message, - channel=task.channel, - source=task.source, - username=task.username, - original_chat_id=task.original_chat_id, - reply_mode=( - ReplyMode.DISPATCH - if has_message_context - else ReplyMode.CAPTURE_ONLY - ), - allow_message_tools=has_message_context, + channel=None, + source=None, + username=notification_username, + original_chat_id=None, + reply_mode=ReplyMode.DISPATCH, + allow_message_tools=True, wait_for_completion=True, ) result_text = str(result or "").strip() @@ -2175,47 +2171,21 @@ class AgentManager: result = "定时任务已执行，但 Agent 未返回结果" await AgentChain().async_post_message( Notification( - channel=task.channel if has_message_context else None, - source=task.source if has_message_context else None, mtype=NotificationType.Agent, - userid=task.user_id if has_message_context else None, - username=( - task.username - if has_message_context - else settings.SUPERUSER - ), - original_chat_id=task.original_chat_id, + username=notification_username, title=f"定时任务：{task.name}", text=result, save_history=False, ) ) - elif not has_message_context: - await AgentChain().async_post_message( - Notification( - mtype=NotificationType.Agent, - username=settings.SUPERUSER, - title=f"定时任务：{task.name}", - text=result_text, - save_history=False, - ) - ) except Exception as err: success = False result = f"Agent 定时任务执行失败：{str(err)}" logger.error(f"Agent 定时任务 {task_id} 执行失败: {str(err)}") await AgentChain().async_post_message( Notification( - channel=task.channel if has_message_context else None, - source=task.source if has_message_context else None, mtype=NotificationType.Agent, - userid=task.user_id if has_message_context else None, - username=( - task.username - if has_message_context - else settings.SUPERUSER - ), - original_chat_id=task.original_chat_id, + username=notification_username, title=f"定时任务执行失败：{task.name}", text=result, save_history=False, diff --git a/app/agent/middleware/jobs.py b/app/agent/middleware/jobs.py index ad4a0da1..d542c6a5 100644 --- a/app/agent/middleware/jobs.py +++ b/app/agent/middleware/jobs.py @@ -205,8 +205,11 @@ You have a scheduled jobs system for user-requested delayed or recurring work. Rules: - For new delayed, recurring, reminder, or monitoring work, use the dedicated - `create_agent_task`, `query_agent_tasks`, `update_agent_task`, and - `delete_agent_task` tools. Do not create or edit JOB.md files for new tasks. + `create_agent_task`, `query_agent_tasks`, `update_agent_task`, `run_agent_task`, + and `delete_agent_task` tools. These tools use integer task IDs. Do not create + or edit JOB.md files for new tasks. +- Use `query_schedulers` and `run_scheduler` only for MoviePilot system, plugin, + or workflow runtime services; never pass their string job IDs to Agent task tools. - Do not create tasks for immediate one-time work or work already handled by MoviePilot schedulers. - Entries listed above are legacy JOB.md tasks. Read their files only when a heartbeat asks you to execute them. - During heartbeat checks, act only on `pending` or `in_progress` jobs, update status/last_run/logs, and leave recurring jobs `pending` after each run. diff --git a/app/agent/prompt/System Core Prompt.txt b/app/agent/prompt/System Core Prompt.txt index ac784a38..5bc9ee78 100644 --- a/app/agent/prompt/System Core Prompt.txt +++ b/app/agent/prompt/System Core Prompt.txt @@ -24,7 +24,7 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel - Do not stop for approval on read-only operations. - If the user has not explicitly requested an operation that changes system behavior, ask for confirmation before proceeding. This includes modifying system settings, updating plugin configuration, reloading plugins, running restart/stop/start commands, or triggering slash commands such as `/restart`. - Always get explicit consent before destructive or high-impact actions such as starting downloads, deleting subscriptions, deleting download tasks or files, removing history, installing/uninstalling plugins, changing site authentication, changing scheduler or workflow execution state, restarting services, or stopping services. -- When the user explicitly asks for delayed, recurring, reminder, or monitoring work, use `create_agent_task` instead of promising to remember it or writing a JOB.md file. Use a `date` trigger with `delay_minutes` for requests such as "in 30 minutes", an exact `date` trigger for other single future runs, and a five-field `cron` trigger for recurring work. Manage existing autonomous tasks with `query_agent_tasks`, `update_agent_task`, and `delete_agent_task`. +- When the user explicitly asks for delayed, recurring, reminder, or monitoring work, use `create_agent_task` instead of promising to remember it or writing a JOB.md file. Use a `date` trigger with `delay_minutes` for requests such as "in 30 minutes", an exact `date` trigger for other single future runs, and a five-field `cron` trigger for recurring work. Manage existing autonomous tasks with `query_agent_tasks`, `update_agent_task`, `run_agent_task`, and `delete_agent_task`; these tools use integer `task_id` values. Use `query_schedulers` and `run_scheduler` only for MoviePilot system, plugin, or workflow runtime services, whose string `job_id` values must never be passed to autonomous-task tools. - If the user explicitly requested the exact write action, perform the smallest correct change and then validate the result. - If a requested action is ambiguous between read-only inspection and state change, inspect first and ask a short confirmation question before the state-changing step. diff --git a/app/agent/tools/base.py b/app/agent/tools/base.py index 2667b56b..1e761e04 100644 --- a/app/agent/tools/base.py +++ b/app/agent/tools/base.py @@ -620,7 +620,8 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta): 发送工具通知消息。 WebAgent 渠道没有后端模块实例，前端流式面板通过 Agent 上下文中的 - 回调直接接收通知；其它渠道继续走统一消息链。 + 回调直接接收通知；无渠道的后台任务清空渠道侧定位信息后交由消息链广播， + 其它渠道继续走统一消息链。 """ callback = self._agent_context.get("notification_callback") if ( @@ -630,6 +631,20 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta): callback(notification) return + if not self._channel or not self._source: + notification = notification.model_copy( + update={ + "channel": None, + "source": None, + "userid": None, + "username": notification.username + or self._username + or settings.SUPERUSER, + "original_message_id": None, + "original_chat_id": None, + } + ) + await ToolChain().async_post_message(notification) async def send_tool_message( diff --git a/app/agent/tools/factory.py b/app/agent/tools/factory.py index 89aca6af..338e6fb7 100644 --- a/app/agent/tools/factory.py +++ b/app/agent/tools/factory.py @@ -46,6 +46,7 @@ from app.agent.tools.impl.create_agent_task import CreateAgentTaskTool from app.agent.tools.impl.delete_agent_task import DeleteAgentTaskTool from app.agent.tools.impl.query_agent_tasks import QueryAgentTasksTool from app.agent.tools.impl.query_schedulers import QuerySchedulersTool +from app.agent.tools.impl.run_agent_task import RunAgentTaskTool from app.agent.tools.impl.run_scheduler import RunSchedulerTool from app.agent.tools.impl.update_agent_task import UpdateAgentTaskTool from app.agent.tools.impl.query_workflows import QueryWorkflowsTool @@ -148,6 +149,7 @@ class MoviePilotToolFactory: CreateAgentTaskTool, QueryAgentTasksTool, UpdateAgentTaskTool, + RunAgentTaskTool, DeleteAgentTaskTool, QuerySchedulersTool, RunSchedulerTool, diff --git a/app/agent/tools/impl/create_agent_task.py b/app/agent/tools/impl/create_agent_task.py index ea373081..9a40ff7b 100644 --- a/app/agent/tools/impl/create_agent_task.py +++ b/app/agent/tools/impl/create_agent_task.py @@ -81,14 +81,14 @@ class CreateAgentTaskTool(MoviePilotTool): """创建可精确唤醒当前 Agent 会话的自主定时任务。""" name: str = "create_agent_task" - tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, ToolTag.Admin] + tags: list[str] = [ToolTag.Write, ToolTag.AgentTask, ToolTag.Admin] description: str = ( "Create a persistent autonomous agent task only when the user explicitly asks " "for delayed, scheduled, recurring, reminder, or monitoring work. Use trigger_type " "'date' with delay_minutes for requests such as 'check in 30 minutes', an exact " "trigger time for other one-time work, and 'cron' for recurring schedules. When " "fired, MoviePilot wakes the agent in this conversation, executes content, and " - "sends the result to the user." + "broadcasts user-facing messages through the configured notification channels." ) args_schema: Type[BaseModel] = CreateAgentTaskInput require_admin: bool = True diff --git a/app/agent/tools/impl/delete_agent_task.py b/app/agent/tools/impl/delete_agent_task.py index 70418088..411b0da1 100644 --- a/app/agent/tools/impl/delete_agent_task.py +++ b/app/agent/tools/impl/delete_agent_task.py @@ -17,7 +17,7 @@ class DeleteAgentTaskTool(MoviePilotTool): """永久删除 Agent 自主定时任务。""" name: str = "delete_agent_task" - tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, ToolTag.Admin] + tags: list[str] = [ToolTag.Write, ToolTag.AgentTask, ToolTag.Admin] description: str = ( "Permanently delete an autonomous agent task and remove its runtime schedule. " "Use update_agent_task with enabled=false when the user only wants to pause it." diff --git a/app/agent/tools/impl/query_agent_tasks.py b/app/agent/tools/impl/query_agent_tasks.py index 9674b0c9..16e88c9a 100644 --- a/app/agent/tools/impl/query_agent_tasks.py +++ b/app/agent/tools/impl/query_agent_tasks.py @@ -27,10 +27,12 @@ class QueryAgentTasksTool(MoviePilotTool): """查询当前用户创建的 Agent 自主定时任务。""" name: str = "query_agent_tasks" - tags: list[str] = [ToolTag.Read, ToolTag.Scheduler, ToolTag.Admin] + tags: list[str] = [ToolTag.Read, ToolTag.AgentTask, ToolTag.Admin] description: str = ( - "Query persistent autonomous agent tasks, including their task content, exact " - "date or cron trigger, enabled state, next run time, and latest execution result." + "Query persistent autonomous agent tasks owned by the current user, including " + "reminders, monitoring tasks, and recurring agent work. Returns the integer " + "task_id, instructions, trigger, enabled state, next run time, and latest result. " + "Do not use this for MoviePilot system, plugin, or workflow scheduler services." ) args_schema: Type[BaseModel] = QueryAgentTasksInput require_admin: bool = True diff --git a/app/agent/tools/impl/query_schedulers.py b/app/agent/tools/impl/query_schedulers.py index b73249a8..43b4f063 100644 --- a/app/agent/tools/impl/query_schedulers.py +++ b/app/agent/tools/impl/query_schedulers.py @@ -3,7 +3,7 @@ import json from typing import Optional, Type -from pydantic import BaseModel, Field +from pydantic import BaseModel from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag @@ -11,47 +11,69 @@ from app.log import logger class QuerySchedulersInput(BaseModel): - """查询定时服务工具的输入参数模型""" + """查询运行时定时服务的输入参数模型。""" + + class QuerySchedulersTool(MoviePilotTool): + """查询系统、插件和工作流注册的运行时定时服务。""" + name: str = "query_schedulers" tags: list[str] = [ ToolTag.Read, ToolTag.Scheduler, + ToolTag.Admin, ] - description: str = "Query scheduled tasks and list all available scheduler jobs. Shows job status, next run time, and provider information." + description: str = ( + "Query runtime scheduler services registered by MoviePilot system components, " + "plugins, and workflows. It excludes user-created autonomous agent tasks; use " + "query_agent_tasks for reminders, monitoring tasks, and other agent schedules." + ) args_schema: Type[BaseModel] = QuerySchedulersInput + require_admin: bool = True - def get_tool_message(self, **kwargs) -> Optional[str]: - """生成友好的提示消息""" - return "查询定时服务" + def get_tool_message(self, **kwargs: object) -> Optional[str]: + """生成查询运行时定时服务的提示消息。""" + return "查询系统定时服务" - async def run(self, **kwargs) -> str: + async def run(self, **kwargs: object) -> str: + """查询非 Agent 自主任务的运行时定时服务。""" logger.info(f"执行工具: {self.name}") try: - from app.scheduler import Scheduler + from app.scheduler import AGENT_TASK_JOB_PREFIX, Scheduler scheduler = Scheduler() - schedulers = scheduler.list() + agent_task_prefix = f"{AGENT_TASK_JOB_PREFIX}-" + schedulers = [ + scheduler_item + for scheduler_item in scheduler.list() + if not str(scheduler_item.id or "").startswith(agent_task_prefix) + ] if schedulers: - # 转换为字典列表以便JSON序列化 - schedulers_list = [] - for s in schedulers: - schedulers_list.append({ - "id": s.id, - "name": s.name, - "provider": s.provider, - "status": s.status, - "next_run": s.next_run - }) + schedulers_list = [ + { + "id": scheduler_item.id, + "name": scheduler_item.name, + "provider": scheduler_item.provider, + "status": scheduler_item.status, + "next_run": scheduler_item.next_run, + } + for scheduler_item in schedulers + ] result_json = json.dumps(schedulers_list, ensure_ascii=False, indent=2) - # 限制最多30条结果 total_count = len(schedulers_list) if total_count > 30: limited_schedulers = schedulers_list[:30] - limited_json = json.dumps(limited_schedulers, ensure_ascii=False, indent=2) - return f"注意：查询结果共找到 {total_count} 条，为节省上下文空间，仅显示前 30 条结果。\n\n{limited_json}" + limited_json = json.dumps( + limited_schedulers, + ensure_ascii=False, + indent=2, + ) + return ( + f"注意：查询结果共找到 {total_count} 条，为节省上下文空间，" + f"仅显示前 30 条结果。\n\n{limited_json}" + ) return result_json - return "未找到定时服务" + return "未找到系统、插件或工作流定时服务" except Exception as e: logger.error(f"查询定时服务失败: {e}", exc_info=True) return f"查询定时服务时发生错误: {str(e)}" diff --git a/app/agent/tools/impl/run_agent_task.py b/app/agent/tools/impl/run_agent_task.py new file mode 100644 index 00000000..9a11f4f8 --- /dev/null +++ b/app/agent/tools/impl/run_agent_task.py @@ -0,0 +1,78 @@ +"""立即执行 Agent 自主定时任务工具。""" + +from typing import Optional, Type + +from pydantic import BaseModel, Field + +from app.agent.tools.base import MoviePilotTool +from app.agent.tools.tags import ToolTag +from app.db.agenttask_oper import AgentTaskOper + + +class RunAgentTaskInput(BaseModel): + """立即执行 Agent 自主定时任务的输入参数。""" + + task_id: int = Field( + ..., + ge=1, + description=( + "Integer autonomous task ID returned by query_agent_tasks. Do not pass a " + "runtime scheduler job_id such as agent-task-12." + ), + ) + + +class RunAgentTaskTool(MoviePilotTool): + """将当前用户的 Agent 自主定时任务提交为立即执行。""" + + name: str = "run_agent_task" + tags: list[str] = [ToolTag.Write, ToolTag.AgentTask, ToolTag.Admin] + description: str = ( + "Queue an enabled autonomous agent task owned by the current user for immediate " + "execution. Use the integer task_id returned by query_agent_tasks. The task runs " + "after the current agent turn can finish and broadcasts its result through the " + "configured notification channels." + ) + args_schema: Type[BaseModel] = RunAgentTaskInput + require_admin: bool = True + + def get_tool_message(self, **kwargs: object) -> Optional[str]: + """生成立即执行 Agent 任务的提示消息。""" + return f"立即执行自主定时任务：{kwargs.get('task_id', '')}" + + def _get_task_state(self, task_id: int) -> tuple[str, Optional[str]]: + """校验任务归属和状态，返回可执行性及任务名称。""" + task = AgentTaskOper().get( + task_id=task_id, + user_id=str(self._user_id), + ) + if not task: + return "not_found", None + if not task.enabled: + return "disabled", task.name + if task.last_status == "running": + return "running", task.name + return "ready", task.name + + async def run(self, task_id: int, **kwargs: object) -> str: + """立即执行当前用户拥有且已启用的 Agent 自主定时任务。""" + from app.scheduler import Scheduler + + payload = RunAgentTaskInput(task_id=task_id) + status, task_name = await self.run_blocking( + "db", + self._get_task_state, + payload.task_id, + ) + if status == "not_found": + return f"Agent 定时任务 {task_id} 不存在或不属于当前用户" + if status == "disabled": + return f"Agent 定时任务 {task_id} 已暂停，请先恢复后再执行" + if status == "running": + return f"Agent 定时任务 {task_id} 正在执行，请勿重复触发" + if not Scheduler().start_agent_task(payload.task_id): + return f"Agent 定时任务 {task_id} 尚未注册到运行时调度器，无法立即执行" + return ( + f"Agent 定时任务 {task_id} 已提交立即执行：{task_name}。" + "执行完成后将通过已配置的通知渠道广播结果" + ) diff --git a/app/agent/tools/impl/run_scheduler.py b/app/agent/tools/impl/run_scheduler.py index 040b2431..ce10981d 100644 --- a/app/agent/tools/impl/run_scheduler.py +++ b/app/agent/tools/impl/run_scheduler.py @@ -14,23 +14,32 @@ class RunSchedulerInput(BaseModel): job_id: str = Field( ..., - description="The ID of the scheduled job to run (can be obtained from query_schedulers tool)", + description=( + "Runtime scheduler job ID returned by query_schedulers. Do not pass an " + "autonomous agent task ID or an agent-task-* runtime ID." + ), ) class RunSchedulerTool(MoviePilotTool): + """立即运行系统、插件或工作流注册的定时服务。""" + name: str = "run_scheduler" tags: list[str] = [ ToolTag.Write, ToolTag.Scheduler, ToolTag.Admin, ] - description: str = "Manually trigger a scheduled task to run immediately. This will execute the specified scheduler job by its ID." + description: str = ( + "Manually trigger a MoviePilot system, plugin, or workflow scheduler service by " + "the runtime job_id returned from query_schedulers. This tool does not run " + "user-created autonomous agent tasks; use run_agent_task with an integer task_id." + ) args_schema: Type[BaseModel] = RunSchedulerInput require_admin: bool = True - def get_tool_message(self, **kwargs) -> Optional[str]: - """根据运行参数生成友好的提示消息""" + def get_tool_message(self, **kwargs: object) -> Optional[str]: + """根据运行参数生成友好的提示消息。""" job_id = kwargs.get("job_id", "") return f"运行定时服务 (ID: {job_id})" @@ -46,10 +55,18 @@ class RunSchedulerTool(MoviePilotTool): return True, scheduler_item.name return False, "" - async def run(self, job_id: str, **kwargs) -> str: + async def run(self, job_id: str, **kwargs: object) -> str: + """立即运行非 Agent 自主任务的运行时定时服务。""" logger.info(f"执行工具: {self.name}, 参数: job_id={job_id}") try: + from app.scheduler import AGENT_TASK_JOB_PREFIX + + if job_id.startswith(f"{AGENT_TASK_JOB_PREFIX}-"): + return ( + "Agent 自主定时任务不能通过 run_scheduler 运行，" + "请使用 query_agent_tasks 查询整数 task_id 后调用 run_agent_task" + ) job_exists, job_name = await self.run_blocking( "workflow", self._run_scheduler_sync, job_id ) diff --git a/app/agent/tools/impl/update_agent_task.py b/app/agent/tools/impl/update_agent_task.py index 72d196df..7440aa37 100644 --- a/app/agent/tools/impl/update_agent_task.py +++ b/app/agent/tools/impl/update_agent_task.py @@ -85,7 +85,7 @@ class UpdateAgentTaskTool(MoviePilotTool): """修改、暂停或恢复 Agent 自主定时任务。""" name: str = "update_agent_task" - tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, ToolTag.Admin] + tags: list[str] = [ToolTag.Write, ToolTag.AgentTask, ToolTag.Admin] description: str = ( "Update an autonomous agent task's name, instructions, exact date or cron " "trigger, relative delay_minutes, or enabled state. Use enabled=false to pause " diff --git a/app/agent/tools/tags.py b/app/agent/tools/tags.py index ed028026..6538e27f 100644 --- a/app/agent/tools/tags.py +++ b/app/agent/tools/tags.py @@ -25,6 +25,7 @@ class ToolTag(str, Enum): Plugin = "plugin" Workflow = "workflow" Scheduler = "scheduler" + AgentTask = "agent_task" File = "file" Directory = "directory" Web = "web" diff --git a/app/api/endpoints/agent.py b/app/api/endpoints/agent.py index 5933764a..560f99b1 100644 --- a/app/api/endpoints/agent.py +++ b/app/api/endpoints/agent.py @@ -219,7 +219,9 @@ class _WebAgentMoviePilotAgent(MoviePilotAgent): self.stream_handler = _WebAgentStreamingHandler(self._emit_output) def _should_stream(self) -> bool: - """Web 面板需要实时输出，即使 Web 渠道本身不支持消息编辑。""" + """Web 对话实时输出，复用会话执行后台任务时改用非流式广播。""" + if self.is_background: + return False return True def set_notification_callback( diff --git a/app/scheduler.py b/app/scheduler.py index 44de0ef1..9a787ed7 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -991,6 +991,21 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """生成 Agent 自主定时任务的调度器 Job ID。""" return f"{AGENT_TASK_JOB_PREFIX}-{task_id}" + def start_agent_task(self, task_id: int) -> bool: + """ + 将指定 Agent 自主定时任务提交到运行时调度器立即执行。 + + :param task_id: Agent 自主定时任务 ID + :return: 任务存在且未运行时返回 True，否则返回 False + """ + job_id = self._get_agent_task_job_id(task_id) + with self._lock: + job = self._jobs.get(job_id) + if not job or job.get("running"): + return False + self.start(job_id) + return True + def init_agent_task_jobs(self) -> None: """ 从数据库恢复所有启用的 Agent 自主定时任务。 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index cd718231..844c3bc8 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -187,12 +187,17 @@ FastAPI 异常响应保留 `detail` 字段，并在错误详情为文本时返 | 工具 | 说明 | | :--- | :--- | -| `create_agent_task` | 创建单次或周期任务，并保存任务内容及当前用户、会话和消息渠道 | +| `create_agent_task` | 创建单次或周期任务，并保存任务内容及当前用户、会话上下文 | | `query_agent_tasks` | 查询任务配置、启用状态、下次执行时间及最近执行结果 | | `update_agent_task` | 修改任务内容或触发器，也可通过 `enabled` 暂停、恢复任务 | +| `run_agent_task` | 使用整数 `task_id` 将当前用户已启用的任务提交为立即执行 | | `delete_agent_task` | 永久删除任务并立即移除运行时调度 | -`trigger_type=date` 表示单次执行：“30 分钟后检查”这类相对时间传 `delay_minutes=30`，由后端计算精确时间；固定时间则传 ISO 8601 `trigger`，支持精确到秒。`trigger_type=cron` 使用标准五段 cron（分、时、日、月、周），适合周期检查。未显式携带时区的时间按 MoviePilot 的 `TZ` 配置解释。任务由内存调度器精确触发，配置持久化到数据库，服务重启后会自动恢复；触发后 Agent 在原会话中执行 `content` 并把结果发送到原消息渠道。通过无会话上下文的 MCP/CLI 创建时，结果改发到 MoviePilot 已配置的管理员通知渠道。 +`trigger_type=date` 表示单次执行：“30 分钟后检查”这类相对时间传 `delay_minutes=30`，由后端计算精确时间；固定时间则传 ISO 8601 `trigger`，支持精确到秒。`trigger_type=cron` 使用标准五段 cron（分、时、日、月、周），适合周期检查。未显式携带时区的时间按 MoviePilot 的 `TZ` 配置解释。任务由内存调度器精确触发，配置持久化到数据库，服务重启后会自动恢复；触发后 Agent 在原会话中执行 `content`，执行过程及最终结果均不绑定创建任务时的消息渠道，而是通过 MoviePilot 已配置的通知渠道广播。如果 Agent 在执行过程中已通过消息工具发送完整结果，任务结束时不会再次发送相同的最终回复。 + +Agent 自主任务工具使用数据库中的整数 `task_id`。`query_schedulers` 与 `run_scheduler` 仅面向系统、插件和工作流注册的运行时定时服务，使用字符串 `job_id`，不会返回或执行 `agent-task-*`。两类 ID 不可混用；需要立即执行自主任务时，应先通过 `query_agent_tasks` 确认归属和状态，再调用 `run_agent_task`。立即执行只提交任务，不在当前工具调用内等待结果，从而避免同一 Agent 会话互相等待；执行结果仍按上述通知规则广播。 + +上述过滤只约束 Agent 工具，避免模型混用两类任务。前端系统设置和仪表盘使用的 `/api/v1/dashboard/schedule` 仍返回完整运行时列表，其中包含 `provider=[Agent]` 的自主任务；前端通过 `/api/v1/system/runscheduler` 立即执行这类列表项的行为也保持不变。 创建单次任务的参数示例： diff --git a/skills/moviepilot-cli/SKILL.md b/skills/moviepilot-cli/SKILL.md index b4faf42e..d5d01962 100644 --- a/skills/moviepilot-cli/SKILL.md +++ b/skills/moviepilot-cli/SKILL.md @@ -1,6 +1,6 @@ --- name: moviepilot-cli -version: 4 +version: 6 description: >- Use this skill when the user asks to operate MoviePilot through the local `moviepilot tool` MCP CLI for normal product workflows: media search, torrent @@ -58,7 +58,7 @@ Always run `show ` before calling a command — parameter names are not | Library | query_library_exists, query_library_latest, transfer_file, scrape_metadata, query_transfer_history | | Files | list_directory, query_directory_settings | | Sites | query_sites, query_site_userdata, test_site, update_site, update_site_cookie | -| System | query_schedulers, run_scheduler, create_agent_task, query_agent_tasks, update_agent_task, delete_agent_task, query_workflows, run_workflow, query_rule_groups, query_episode_schedule, send_message | +| System | query_schedulers, run_scheduler, create_agent_task, query_agent_tasks, update_agent_task, run_agent_task, delete_agent_task, query_workflows, run_workflow, query_rule_groups, query_episode_schedule, send_message | ## Workflows @@ -192,6 +192,17 @@ Use autonomous tasks only when the user explicitly requests delayed, recurring, reminder, or monitoring behavior. Immediate work should run directly. Use the MoviePilot `TZ` setting for local times. +Scheduled runs reuse the original Agent session context, but user-facing +messages are broadcast through MoviePilot's configured notification channels +instead of being tied to the channel that created the task. If the Agent sends +the complete result with a message tool during execution, it does not send the +same final reply again when the run finishes. + +Autonomous task tools use the integer `task_id` returned by +`query_agent_tasks`. `query_schedulers` and `run_scheduler` are only for +MoviePilot system, plugin, and workflow runtime services and use string +`job_id` values; never mix these IDs or use those tools for autonomous tasks. + For a relative one-time request, use `date` with `delay_minutes`; MoviePilot calculates and persists the exact run time: `moviepilot tool run create_agent_task name="检查电影资源" content="搜索电影《示例电影》是否有资源并报告，不要自动下载。" trigger_type=date delay_minutes=30` @@ -209,6 +220,10 @@ List tasks and inspect `next_run_at` and the latest result: Pause or resume a task: `moviepilot tool run update_agent_task task_id=1 enabled=false` +Queue an enabled task for immediate execution without waiting in the current +Agent turn: +`moviepilot tool run run_agent_task task_id=1` + Delete a task only after confirming permanent removal with the user: `moviepilot tool run delete_agent_task task_id=1` diff --git a/tests/test_agent_scheduled_tasks.py b/tests/test_agent_scheduled_tasks.py index dc9ddcbe..eb87a939 100644 --- a/tests/test_agent_scheduled_tasks.py +++ b/tests/test_agent_scheduled_tasks.py @@ -1,14 +1,24 @@ import json import threading from datetime import datetime, timedelta -from unittest.mock import AsyncMock +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock, patch from uuid import uuid4 import pytest import pytz from apscheduler.schedulers.background import BackgroundScheduler +from langchain_core.messages import AIMessage -from app.agent import AgentChain, AgentManager, ReplyMode +from app.agent import ( + AgentChain, + AgentManager, + MoviePilotAgent, + ReplyMode, + _MessageTask, +) +from app.agent.middleware.tool_selection import ToolSelectorMiddleware from app.agent.tools.factory import MoviePilotToolFactory from app.agent.tools.impl.create_agent_task import ( CreateAgentTaskInput, @@ -16,9 +26,15 @@ from app.agent.tools.impl.create_agent_task import ( ) from app.agent.tools.impl.delete_agent_task import DeleteAgentTaskTool from app.agent.tools.impl.query_agent_tasks import QueryAgentTasksTool +from app.agent.tools.impl.query_schedulers import QuerySchedulersTool +from app.agent.tools.impl.run_agent_task import RunAgentTaskTool +from app.agent.tools.impl.run_scheduler import RunSchedulerTool +from app.agent.tools.impl.send_message import SendMessageTool from app.agent.tools.impl.update_agent_task import UpdateAgentTaskTool +from app.agent.tools.tags import ToolTag from app.core.config import settings from app.db.agenttask_oper import AgentTaskOper +from app.schemas import ScheduleInfo from app.scheduler import Scheduler from app.utils.timer import TimerUtils @@ -30,6 +46,7 @@ class _FakeAgentTaskScheduler: """初始化运行时调度记录。""" self.updated = [] self.removed = [] + self.started = [] def update_agent_task_job(self, task_id: int) -> str: """记录任务重载并返回固定的下一次执行时间。""" @@ -44,6 +61,11 @@ class _FakeAgentTaskScheduler: """返回固定的下一次执行时间。""" return "2099-01-01T00:00:00+08:00" + def start_agent_task(self, task_id: int) -> bool: + """记录 Agent 任务立即执行投递。""" + self.started.append(task_id) + return True + @pytest.fixture def anyio_backend() -> str: @@ -114,10 +136,81 @@ def test_agent_task_tools_are_registered_with_relative_delay_schema() -> None: "create_agent_task", "query_agent_tasks", "update_agent_task", + "run_agent_task", "delete_agent_task", }.issubset(tool_names) + agent_task_tool_names = { + "create_agent_task", + "query_agent_tasks", + "update_agent_task", + "run_agent_task", + "delete_agent_task", + } + assert [ + tool_class.model_fields["name"].default + for tool_class in MoviePilotToolFactory.BUILTIN_TOOL_CLASSES + if tool_class.model_fields["name"].default in agent_task_tool_names + ] == [ + "create_agent_task", + "query_agent_tasks", + "update_agent_task", + "run_agent_task", + "delete_agent_task", + ] assert "delay_minutes" in CreateAgentTaskInput.model_json_schema()["properties"] + agent_task_tools = [ + _build_tool(tool_class, "admin-user") + for tool_class in ( + CreateAgentTaskTool, + QueryAgentTasksTool, + UpdateAgentTaskTool, + RunAgentTaskTool, + DeleteAgentTaskTool, + ) + ] + assert all(ToolTag.AgentTask.value in tool.tags for tool in agent_task_tools) + assert all(ToolTag.Scheduler.value not in tool.tags for tool in agent_task_tools) + scheduler_tool = _build_tool(QuerySchedulersTool, "admin-user") + assert ToolTag.Scheduler.value in scheduler_tool.tags + assert ToolTag.AgentTask.value not in scheduler_tool.tags + runtime_scheduler_tools = [ + scheduler_tool, + _build_tool(RunSchedulerTool, "admin-user"), + ] + all_scheduler_tools = [*agent_task_tools, *runtime_scheduler_tools] + tool_groups = dict( + ToolSelectorMiddleware._build_tool_groups( + available_tools=all_scheduler_tools, + valid_tool_names=[tool.name for tool in all_scheduler_tools], + ) + ) + assert tool_groups[ToolTag.AgentTask.value] == [ + "create_agent_task", + "query_agent_tasks", + "update_agent_task", + "run_agent_task", + "delete_agent_task", + ] + assert tool_groups[ToolTag.Scheduler.value] == [ + "query_schedulers", + "run_scheduler", + ] + + +def test_agent_prompt_declares_scheduler_tool_boundaries() -> None: + """核心提示应明确自主任务与运行时调度服务的工具和 ID 边界。""" + project_root = Path(__file__).resolve().parents[1] + core_prompt = (project_root / "app/agent/prompt/System Core Prompt.txt").read_text( + encoding="utf-8" + ) + + assert "Manage existing autonomous tasks with `query_agent_tasks`" in core_prompt + assert "`run_agent_task`" in core_prompt + assert "Use `query_schedulers` and `run_scheduler` only" in core_prompt + assert "integer `task_id`" in core_prompt + assert "string `job_id`" in core_prompt + def test_agent_task_oper_persists_and_scopes_tasks() -> None: """AgentTaskOper 应持久化任务并按创建用户隔离查询和修改。""" @@ -155,7 +248,7 @@ def test_agent_task_oper_persists_and_scopes_tasks() -> None: def test_scheduler_registers_and_removes_agent_task_job() -> None: - """Scheduler 应把数据库任务注册为精确 APScheduler Job 并可动态移除。""" + """Scheduler 应注册并完整列出 Agent 任务，供前端展示和动态移除。""" user_id = f"user-{uuid4().hex}" task = AgentTaskOper().add( name="定时检查", @@ -181,10 +274,111 @@ def test_scheduler_registers_and_removes_agent_task_job() -> None: assert next_run_at assert scheduler._scheduler.get_job(job_id) is not None assert scheduler._jobs[job_id]["kwargs"] == {"task_id": task.id} + scheduler._scheduler.start(paused=True) + try: + scheduler_items = scheduler.list() + agent_scheduler_item = next( + item for item in scheduler_items if item.id == job_id + ) + assert agent_scheduler_item.name == task.name + assert agent_scheduler_item.provider == "[Agent]" - scheduler.remove_agent_task_job(task.id) - assert scheduler._scheduler.get_job(job_id) is None - assert job_id not in scheduler._jobs + scheduler.remove_agent_task_job(task.id) + assert scheduler._scheduler.get_job(job_id) is None + assert job_id not in scheduler._jobs + finally: + scheduler._scheduler.shutdown(wait=False) + + +def test_scheduler_starts_registered_agent_task_without_waiting() -> None: + """Agent 任务立即执行入口应只调用运行时 start，并拒绝缺失或运行中的任务。""" + scheduler = object.__new__(Scheduler) + scheduler._lock = threading.RLock() + scheduler._jobs = { + "agent-task-7": { + "name": "测试 Agent 任务", + "running": False, + } + } + scheduler.start = Mock() + + assert scheduler.start_agent_task(7) is True + scheduler.start.assert_called_once_with("agent-task-7") + + scheduler._jobs["agent-task-7"]["running"] = True + assert scheduler.start_agent_task(7) is False + assert scheduler.start_agent_task(8) is False + + +@pytest.mark.anyio +async def test_dashboard_schedule_keeps_agent_tasks(monkeypatch) -> None: + """前端后台服务接口必须保留 Agent 自主任务。""" + from app.api.endpoints.dashboard import schedule + + scheduler_items = [ + ScheduleInfo( + id="agent-task-7", + name="检查资源", + provider="[Agent]", + status="等待", + next_run="20 分钟后", + ) + ] + monkeypatch.setattr( + "app.api.endpoints.dashboard.Scheduler", + lambda: SimpleNamespace(list=lambda: scheduler_items), + ) + + result = await schedule(None) + + assert result == scheduler_items + assert result[0].id == "agent-task-7" + assert result[0].provider == "[Agent]" + + +@pytest.mark.anyio +async def test_scheduler_tools_exclude_agent_tasks(monkeypatch) -> None: + """运行时调度查询应过滤 Agent 任务，并把两类查询边界写入工具描述。""" + scheduler = SimpleNamespace( + list=lambda: [ + ScheduleInfo( + id="subscribe_search_all", + name="订阅搜索", + provider="[系统]", + status="等待", + next_run="10 分钟后", + ), + ScheduleInfo( + id="agent-task-7", + name="检查资源", + provider="[Agent]", + status="等待", + next_run="20 分钟后", + ), + ] + ) + monkeypatch.setattr("app.scheduler.Scheduler", lambda: scheduler) + tool = _build_tool(QuerySchedulersTool, "admin-user") + + result = json.loads(await tool.run()) + + assert [item["id"] for item in result] == ["subscribe_search_all"] + assert tool.require_admin is True + assert "excludes user-created autonomous agent tasks" in tool.description + agent_query_tool = _build_tool(QueryAgentTasksTool, "admin-user") + assert "owned by the current user" in agent_query_tool.description + + +@pytest.mark.anyio +async def test_run_scheduler_rejects_agent_task_job_id() -> None: + """run_scheduler 不应接受 Agent 任务的运行时 job_id。""" + tool = _build_tool(RunSchedulerTool, "admin-user") + tool._run_scheduler_sync = Mock() + + result = await tool.run(job_id="agent-task-7") + + assert "run_agent_task" in result + tool._run_scheduler_sync.assert_not_called() @pytest.mark.anyio @@ -243,14 +437,60 @@ async def test_agent_task_tools_manage_persistent_schedule(monkeypatch) -> None: assert running_update == f"Agent 定时任务 {task_id} 正在执行，请稍后再修改" AgentTaskOper().finish(task_id, success=True, result="完成") + run_result = await _build_tool(RunAgentTaskTool, user_id).run(task_id=task_id) + assert f"Agent 定时任务 {task_id} 已提交立即执行" in run_result + assert fake_scheduler.started == [task_id] + deleted = await _build_tool(DeleteAgentTaskTool, user_id).run(task_id=task_id) assert deleted == f"Agent 定时任务 {task_id} 已删除" assert fake_scheduler.removed == [task_id] @pytest.mark.anyio -async def test_agent_manager_executes_task_in_original_session() -> None: - """定时触发应复用原 Agent 会话和渠道，并在单次执行后停用任务。""" +async def test_run_agent_task_enforces_owner_and_enabled_state(monkeypatch) -> None: + """立即执行 Agent 任务时应校验当前用户归属和启用状态。""" + owner_id = f"owner-{uuid4().hex}" + task = AgentTaskOper().add( + name="检查资源", + content="检查资源并报告", + trigger_type="cron", + cron_expression="0 * * * *", + run_at=None, + user_id=owner_id, + username="admin", + session_id=f"session-{owner_id}", + channel="Telegram", + source="telegram-test", + original_chat_id="chat-1", + ) + fake_scheduler = _FakeAgentTaskScheduler() + monkeypatch.setattr("app.scheduler.Scheduler", lambda: fake_scheduler) + + other_user_result = await _build_tool( + RunAgentTaskTool, + "another-user", + ).run(task_id=task.id) + assert "不存在或不属于当前用户" in other_user_result + assert fake_scheduler.started == [] + + AgentTaskOper().update( + task_id=task.id, + user_id=owner_id, + payload={"enabled": False}, + ) + disabled_result = await _build_tool( + RunAgentTaskTool, + owner_id, + ).run(task_id=task.id) + assert "已暂停" in disabled_result + assert fake_scheduler.started == [] + + +@pytest.mark.anyio +async def test_agent_manager_executes_task_with_broadcast_delivery( + monkeypatch, +) -> None: + """定时触发应复用原 Agent 会话、广播结果并在单次执行后停用任务。""" user_id = f"user-{uuid4().hex}" task = AgentTaskOper().add( name="检查电影资源", @@ -267,6 +507,8 @@ async def test_agent_manager_executes_task_in_original_session() -> None: ) manager = AgentManager() manager.process_message = AsyncMock(return_value="已找到 2 个资源") + post_message = AsyncMock() + monkeypatch.setattr(AgentChain, "async_post_message", post_message) success, result = await manager.execute_scheduled_task(task.id) @@ -275,12 +517,14 @@ async def test_agent_manager_executes_task_in_original_session() -> None: kwargs = manager.process_message.await_args.kwargs assert kwargs["session_id"] == task.session_id assert kwargs["user_id"] == user_id - assert kwargs["channel"] == "Telegram" - assert kwargs["source"] == "telegram-test" - assert kwargs["original_chat_id"] == "chat-123" + assert kwargs["channel"] is None + assert kwargs["source"] is None + assert kwargs["original_chat_id"] is None assert kwargs["reply_mode"] == ReplyMode.DISPATCH + assert kwargs["allow_message_tools"] is True assert kwargs["wait_for_completion"] is True assert "搜索示例电影是否已有资源" in kwargs["message"] + post_message.assert_not_awaited() completed = AgentTaskOper().get(task.id) assert completed.enabled is False @@ -312,10 +556,10 @@ async def test_agent_manager_executes_task_in_original_session() -> None: @pytest.mark.anyio -async def test_agent_manager_dispatches_contextless_result_to_admin( - monkeypatch, +async def test_agent_manager_runs_contextless_task_in_broadcast_mode( + monkeypatch, ) -> None: - """无原消息渠道的 MCP 任务应捕获结果并发送到管理员通知渠道。""" + """无原消息渠道的任务也应由 Agent 广播结果且不由调度器重复补发。""" user_id = f"api-{uuid4().hex}" task = AgentTaskOper().add( name="后台检查资源", @@ -340,9 +584,163 @@ async def test_agent_manager_dispatches_contextless_result_to_admin( assert success is True assert result == "后台检查完成" kwargs = manager.process_message.await_args.kwargs - assert kwargs["reply_mode"] == ReplyMode.CAPTURE_ONLY - assert kwargs["allow_message_tools"] is False + assert kwargs["channel"] is None + assert kwargs["source"] is None + assert kwargs["reply_mode"] == ReplyMode.DISPATCH + assert kwargs["allow_message_tools"] is True + post_message.assert_not_awaited() + + +@pytest.mark.anyio +async def test_agent_manager_broadcasts_empty_task_result(monkeypatch) -> None: + """Agent 未返回内容时，调度器应广播一次兜底消息。""" + user_id = f"empty-{uuid4().hex}" + task = AgentTaskOper().add( + name="空结果检查", + content="执行检查", + trigger_type="cron", + cron_expression="0 * * * *", + run_at=None, + user_id=user_id, + username="admin", + session_id=f"session-{user_id}", + channel="Telegram", + source="telegram-test", + original_chat_id="chat-123", + ) + manager = AgentManager() + manager.process_message = AsyncMock(return_value="") + post_message = AsyncMock() + monkeypatch.setattr(AgentChain, "async_post_message", post_message) + + success, result = await manager.execute_scheduled_task(task.id) + + assert success is False + assert result == "定时任务已执行，但 Agent 未返回结果" notification = post_message.await_args.args[0] + assert notification.channel is None + assert notification.source is None assert notification.userid is None - assert notification.username == settings.SUPERUSER - assert notification.text == "后台检查完成" + assert notification.original_chat_id is None + assert notification.username == "admin" + + +@pytest.mark.anyio +async def test_cached_agent_clears_channel_for_background_task() -> None: + """复用会话 Agent 时，后台任务必须覆盖上一轮保留的渠道信息。""" + manager = AgentManager() + agent = MoviePilotAgent( + session_id="scheduled-cached-session", + user_id="user-1", + channel="Telegram", + source="telegram-test", + username="admin", + original_chat_id="chat-123", + ) + agent.process = AsyncMock(return_value="完成") + manager.active_agents[agent.session_id] = agent + task = _MessageTask( + session_id=agent.session_id, + user_id="user-1", + message="执行后台任务", + channel=None, + source=None, + username="admin", + original_chat_id=None, + reply_mode=ReplyMode.DISPATCH, + allow_message_tools=True, + ) + + result = await manager._process_message_internal(task) + + assert result == "完成" + assert agent.channel is None + assert agent.source is None + assert agent.original_chat_id is None + + +@pytest.mark.anyio +async def test_background_agent_final_message_is_broadcast() -> None: + """后台 Agent 的最终消息应清空渠道及渠道用户定位后广播。""" + agent = MoviePilotAgent( + session_id="scheduled-broadcast-session", + user_id="telegram-user-id", + channel=None, + source=None, + username="admin", + original_message_id="message-1", + original_chat_id="chat-1", + ) + + with patch( + "app.agent.AgentChain.async_post_message", + new_callable=AsyncMock, + ) as post_message: + await agent.send_agent_message("任务完成", title="MoviePilot助手") + + notification = post_message.await_args.args[0] + assert notification.channel is None + assert notification.source is None + assert notification.userid is None + assert notification.original_message_id is None + assert notification.original_chat_id is None + assert notification.username == "admin" + + +@pytest.mark.anyio +async def test_background_send_message_tool_broadcasts() -> None: + """后台 send_message 工具应广播消息并记录本轮已经回复。""" + tool = SendMessageTool( + session_id="scheduled-tool-session", + user_id="telegram-user-id", + ) + tool.set_message_attr(channel=None, source=None, username="admin") + agent_context = {} + tool.set_agent_context(agent_context) + + with patch( + "app.agent.tools.base.ToolChain.async_post_message", + new_callable=AsyncMock, + ) as post_message: + result = await tool.run(message="工具已完成任务") + + assert result == "消息已发送" + assert agent_context["user_reply_sent"] is True + notification = post_message.await_args.args[0] + assert notification.channel is None + assert notification.source is None + assert notification.userid is None + assert notification.original_message_id is None + assert notification.original_chat_id is None + assert notification.username == "admin" + + +@pytest.mark.anyio +async def test_background_agent_does_not_repeat_tool_message() -> None: + """消息工具已完成回复后，后台 Agent 不应再次发送最终文本。""" + agent = MoviePilotAgent( + session_id="scheduled-no-repeat-session", + user_id="user-1", + channel=None, + source=None, + username="admin", + replay_mode=ReplyMode.DISPATCH, + ) + 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 + completed_agent = SimpleNamespace( + ainvoke=AsyncMock(return_value=None), + get_state=lambda _config: SimpleNamespace( + values={"messages": [AIMessage(content="消息已发送")]} + ), + ) + agent._create_agent = AsyncMock(return_value=completed_agent) + agent.send_agent_message = AsyncMock() + + await agent._execute_agent([]) + + agent.send_agent_message.assert_not_awaited() diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index 3945e14c..2bdbe845 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -23,7 +23,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: expected_versions = { "database-operation": "3", "moviepilot-api": "2", - "moviepilot-cli": "4", + "moviepilot-cli": "6", "moviepilot-update": "3", } diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 617bac33..fd7e9cf7 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -296,6 +296,21 @@ def test_web_agent_admin_context_uses_current_user_id(): user_oper.return_value.async_get_by_id.assert_awaited_once_with(7) +def test_web_agent_reused_for_background_task_disables_streaming(): + """Web Agent 被后台任务复用且渠道已清空时应改用非流式广播。""" + agent = _WebAgentMoviePilotAgent( + session_id="web-agent:scheduled-session", + user_id="7", + channel=None, + source=None, + username="admin", + replay_mode=ReplyMode.DISPATCH, + ) + + assert agent.is_background is True + assert agent._should_stream() is False + + def test_web_agent_channel_supports_streaming_and_attachments(): """WebAgent 渠道应声明流式、多媒体和文件发送能力。""" assert ChannelCapabilityManager.supports_capability(