Route autonomous agent tasks through notification broadcasting

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,
+                channel=None if broadcast else self.channel,
-                source=self.source,
+                source=None if broadcast else self.source,
                 mtype=NotificationType.Agent,
-                userid=self.user_id,
+                userid=None if broadcast else self.user_id,
-                username=self.username,
+                username=self.username or (settings.SUPERUSER if broadcast else None),
-                original_message_id=self.original_message_id,
+                original_message_id=None if broadcast else self.original_message_id,
-                original_chat_id=self.original_chat_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:
+            # 每条队列任务都携带完整消息上下文，None 也必须覆盖，避免后台任务
-                agent.channel = task.channel
+            # 复用会话 Agent 时继续沿用上一条入站消息的渠道。
-            if task.source:
+            agent.channel = task.channel
-                agent.source = task.source
+            agent.source = task.source
-            if task.username:
+            agent.username = task.username
-                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,
+                channel=None,
-                source=task.source,
+                source=None,
-                username=task.username,
+                username=notification_username,
-                original_chat_id=task.original_chat_id,
+                original_chat_id=None,
-                reply_mode=(
+                reply_mode=ReplyMode.DISPATCH,
-                    ReplyMode.DISPATCH
+                allow_message_tools=True,
-                    if has_message_context
-                    else ReplyMode.CAPTURE_ONLY
-                ),
-                allow_message_tools=has_message_context,
                 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=notification_username,
-                        username=(
-                            task.username
-                            if has_message_context
-                            else settings.SUPERUSER
-                        ),
-                        original_chat_id=task.original_chat_id,
                         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=notification_username,
-                    username=(
-                        task.username
-                        if has_message_context
-                        else settings.SUPERUSER
-                    ),
-                    original_chat_id=task.original_chat_id,
                     title=f"定时任务执行失败：{task.name}",
                     text=result,
                     save_history=False,

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
+  `create_agent_task`, `query_agent_tasks`, `update_agent_task`, `run_agent_task`,
-  `delete_agent_task` tools. Do not create or edit JOB.md files for new tasks.
+  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.

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.
 </confirmation_policy>

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(

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,

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

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."

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 "
+        "Query persistent autonomous agent tasks owned by the current user, including "
-        "date or cron trigger, enabled state, next run time, and latest execution result."
+        "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

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]:
+    def get_tool_message(self, **kwargs: object) -> Optional[str]:
-        """生成友好的提示消息"""
+        """生成查询运行时定时服务的提示消息。"""
-        return "查询定时服务"
+        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 = [
-                schedulers_list = []
+                    {
-                for s in schedulers:
+                        "id": scheduler_item.id,
-                    schedulers_list.append({
+                        "name": scheduler_item.name,
-                        "id": s.id,
+                        "provider": scheduler_item.provider,
-                        "name": s.name,
+                        "status": scheduler_item.status,
-                        "provider": s.provider,
+                        "next_run": scheduler_item.next_run,
-                        "status": s.status,
+                    }
-                        "next_run": s.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)
+                    limited_json = json.dumps(
-                    return f"注意：查询结果共找到 {total_count} 条，为节省上下文空间，仅显示前 30 条结果。\n\n{limited_json}"
+                        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)}"

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}。"
+            "执行完成后将通过已配置的通知渠道广播结果"
+        )

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
             )

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 "

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"

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(

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 自主定时任务。

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` 立即执行这类列表项的行为也保持不变。
 
 创建单次任务的参数示例：
 

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 <command>` 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`
 

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)
+        scheduler.remove_agent_task_job(task.id)
-    assert scheduler._scheduler.get_job(job_id) is None
+        assert scheduler._scheduler.get_job(job_id) is None
-    assert job_id not in scheduler._jobs
+        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:
+async def test_run_agent_task_enforces_owner_and_enabled_state(monkeypatch) -> None:
-    """定时触发应复用原 Agent 会话和渠道，并在单次执行后停用任务。"""
+    """立即执行 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["channel"] is None
-    assert kwargs["source"] == "telegram-test"
+    assert kwargs["source"] is None
-    assert kwargs["original_chat_id"] == "chat-123"
+    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(
+async def test_agent_manager_runs_contextless_task_in_broadcast_mode(
-        monkeypatch,
+    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["channel"] is None
-    assert kwargs["allow_message_tools"] is False
+    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.original_chat_id is None
-    assert notification.text == "后台检查完成"
+    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()

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",
     }
 

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(