diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 34ebed571..7d834ee8b 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -58,6 +58,7 @@ from app.chain import ChainBase from app.core.config import settings from app.core.event import eventmanager from app.db.agentchat_oper import AgentChatOper +from app.db.agenttask_oper import AgentTaskOper from app.db.user_oper import UserOper from app.log import logger from app.schemas import AgentLLMProviderEventData, AgentTokensUsageEventData, Notification, NotificationType @@ -2123,6 +2124,119 @@ class AgentManager: await agent.cleanup() memory_manager.clear_memory(session_id, user_id) + async def execute_scheduled_task(self, task_id: int) -> tuple[bool, str]: + """ + 按持久化上下文唤醒 Agent 执行自主定时任务并向用户回传结果。 + + :param task_id: Agent 定时任务 ID + :return: 执行是否成功及结果摘要 + """ + if not settings.AI_AGENT_ENABLE: + return False, "AI Agent 未启用" + oper = AgentTaskOper() + task = oper.get(task_id) + if not task or not task.enabled: + return False, "Agent 定时任务不存在或已停用" + if not oper.mark_running(task_id): + return False, "Agent 定时任务当前不可执行" + + task_message = ( + f"定时任务已按计划触发。请立即完成下面的任务，不要只确认收到，" + f"也不要重复创建同一个定时任务。\n\n" + f"任务名称：{task.name}\n" + f"任务内容：{task.content}\n\n" + "完成后请直接向用户报告本次执行结果；如果无法完成，请说明原因。" + ) + has_message_context = bool(task.channel and task.source) + success = True + result = "" + 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, + wait_for_completion=True, + ) + result_text = str(result or "").strip() + success = bool(result_text) and not result_text.startswith( + (AGENT_EXECUTION_ERROR_PREFIX, "处理消息时发生错误") + ) + if not result_text: + 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, + 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, + title=f"定时任务执行失败：{task.name}", + text=result, + save_history=False, + ) + ) + finally: + current_task = oper.get(task_id) + oper.finish( + task_id=task_id, + success=success, + result=str(result or ""), + disable=bool( + current_task + and task.trigger_type == "date" + and current_task.trigger_type == task.trigger_type + and current_task.run_at == task.run_at + ), + ) + + return success, str(result or "任务执行完成") + @staticmethod def _build_heartbeat_prompt() -> str: """使用程序内置 System Tasks 定义构建心跳任务提示词。""" diff --git a/app/agent/middleware/jobs.py b/app/agent/middleware/jobs.py index d70190d32..ad4a0da14 100644 --- a/app/agent/middleware/jobs.py +++ b/app/agent/middleware/jobs.py @@ -204,9 +204,11 @@ You have a scheduled jobs system for user-requested delayed or recurring work. {jobs_list} Rules: -- Create jobs only when the user asks for delayed, recurring, reminder, or monitoring behavior. -- Do not create jobs for immediate one-time work or work already handled by MoviePilot schedulers. -- Each job lives in its own directory with a `JOB.md`; read the listed file before executing or updating an active job. +- 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. +- 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. """ @@ -230,7 +232,7 @@ class JobsMiddleware(AgentMiddleware[JobsState, ContextT, ResponseT]): # noqa def _format_jobs_list(jobs: list[JobMetadata]) -> str: """格式化任务元数据列表用于系统提示词。""" if not jobs: - return "(No active jobs. You can create jobs when users request periodic or scheduled tasks.)" + return "(No active legacy JOB.md tasks. Use create_agent_task for new scheduled work.)" lines = [] for job in jobs: diff --git a/app/agent/prompt/System Core Prompt.txt b/app/agent/prompt/System Core Prompt.txt index 732e7b700..ac784a388 100644 --- a/app/agent/prompt/System Core Prompt.txt +++ b/app/agent/prompt/System Core Prompt.txt @@ -24,6 +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`. - 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/factory.py b/app/agent/tools/factory.py index b3b2e7c8c..89aca6af6 100644 --- a/app/agent/tools/factory.py +++ b/app/agent/tools/factory.py @@ -42,8 +42,12 @@ from app.agent.tools.impl.send_message import SendMessageTool from app.agent.tools.impl.ask_user_choice import AskUserChoiceTool from app.agent.tools.impl.send_local_file import SendLocalFileTool from app.agent.tools.impl.send_voice_message import SendVoiceMessageTool +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_scheduler import RunSchedulerTool +from app.agent.tools.impl.update_agent_task import UpdateAgentTaskTool from app.agent.tools.impl.query_workflows import QueryWorkflowsTool from app.agent.tools.impl.run_workflow import RunWorkflowTool from app.agent.tools.impl.query_personas import QueryPersonasTool @@ -141,6 +145,10 @@ class MoviePilotToolFactory: QueryTransferHistoryTool, TransferFileTool, SendMessageTool, + CreateAgentTaskTool, + QueryAgentTasksTool, + UpdateAgentTaskTool, + DeleteAgentTaskTool, QuerySchedulersTool, RunSchedulerTool, QueryWorkflowsTool, @@ -181,6 +189,8 @@ class MoviePilotToolFactory: "edit_file", "execute_command", "ask_user_choice", + "create_agent_task", + "query_agent_tasks", ) @staticmethod diff --git a/app/agent/tools/impl/create_agent_task.py b/app/agent/tools/impl/create_agent_task.py new file mode 100644 index 000000000..9eca1cf21 --- /dev/null +++ b/app/agent/tools/impl/create_agent_task.py @@ -0,0 +1,150 @@ +import json +from datetime import datetime, timedelta +from typing import Literal, Optional, Type + +import pytz +from pydantic import BaseModel, Field, model_validator + +from app.agent.tools.base import MoviePilotTool +from app.agent.tools.tags import ToolTag +from app.core.config import settings +from app.db.agentchat_oper import AgentChatOper +from app.db.agenttask_oper import AgentTaskOper +from app.utils.timer import TimerUtils + + +class CreateAgentTaskInput(BaseModel): + """创建 Agent 自主定时任务的输入参数。""" + + name: str = Field( + ..., + min_length=1, + max_length=100, + description="Short task name shown in task management and execution reports.", + ) + content: str = Field( + ..., + min_length=1, + max_length=10000, + description="Complete instructions that the agent must execute when the task fires.", + ) + trigger_type: Literal["date", "cron"] = Field( + ..., + description="Use 'date' for one exact future run or 'cron' for recurring work.", + ) + trigger: Optional[str] = Field( + None, + min_length=1, + max_length=200, + description=( + "For date, an ISO 8601 local or timezone-aware time such as " + "2026-07-19 20:30:00; for cron, a standard five-field expression " + "(minute hour day month weekday). The MoviePilot system timezone is used." + ), + ) + delay_minutes: Optional[int] = Field( + None, + ge=1, + le=525600, + description=( + "For a one-time date task expressed as 'in N minutes', provide this instead " + "of trigger. MoviePilot calculates and persists the exact future run time." + ), + ) + + @model_validator(mode="after") + def validate_trigger(self) -> "CreateAgentTaskInput": + """校验任务触发配置并统一格式。""" + self.name = self.name.strip() + self.content = self.content.strip() + if not self.name or not self.content: + raise ValueError("name 和 content 不能只包含空白字符") + if self.trigger_type == "date": + if (self.trigger is None) == (self.delay_minutes is None): + raise ValueError("date 任务必须且只能提供 trigger 或 delay_minutes 之一") + if self.delay_minutes is not None: + timezone = pytz.timezone(settings.TZ) + self.trigger = ( + datetime.now(timezone) + timedelta(minutes=self.delay_minutes) + ).isoformat(timespec="seconds") + elif self.trigger is None or self.delay_minutes is not None: + raise ValueError("cron 任务必须提供 trigger，且不能提供 delay_minutes") + self.trigger_type, self.trigger = TimerUtils.normalize_schedule_trigger( + trigger_type=self.trigger_type, + trigger_value=self.trigger, + timezone_name=settings.TZ, + require_future=True, + ) + return self + + +class CreateAgentTaskTool(MoviePilotTool): + """创建可精确唤醒当前 Agent 会话的自主定时任务。""" + + name: str = "create_agent_task" + tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, 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." + ) + args_schema: Type[BaseModel] = CreateAgentTaskInput + require_admin: bool = True + + def get_tool_message(self, **kwargs: object) -> Optional[str]: + """生成创建定时任务的提示消息。""" + return f"创建自主定时任务：{kwargs.get('name', '')}" + + def _create_task(self, payload: CreateAgentTaskInput) -> dict: + """持久化任务并立即注册到运行时调度器。""" + from app.scheduler import Scheduler + + chat = AgentChatOper().get( + session_id=self._session_id, + user_id=self._user_id, + ) + task = AgentTaskOper().add( + name=payload.name.strip(), + content=payload.content.strip(), + trigger_type=payload.trigger_type, + cron_expression=payload.trigger if payload.trigger_type == "cron" else None, + run_at=payload.trigger if payload.trigger_type == "date" else None, + user_id=str(self._user_id), + username=self._username or (chat.username if chat else None), + session_id=str(self._session_id), + channel=self._channel or (chat.channel if chat else None), + source=self._source or (chat.source if chat else None), + original_chat_id=chat.original_chat_id if chat else None, + ) + scheduler = Scheduler() + next_run_at = scheduler.update_agent_task_job(task.id) + return AgentTaskOper.to_dict( + task, + next_run_at=next_run_at, + timezone=settings.TZ, + ) + + async def run( + self, + name: str, + content: str, + trigger_type: str, + trigger: Optional[str] = None, + delay_minutes: Optional[int] = None, + **kwargs: object, + ) -> str: + """创建 Agent 自主定时任务。""" + if not settings.AI_AGENT_ENABLE: + return "AI Agent 未启用，无法创建自主定时任务" + payload = CreateAgentTaskInput( + name=name, + content=content, + trigger_type=trigger_type, + trigger=trigger, + delay_minutes=delay_minutes, + ) + task = await self.run_blocking("db", self._create_task, payload) + return json.dumps(task, ensure_ascii=False, indent=2) diff --git a/app/agent/tools/impl/delete_agent_task.py b/app/agent/tools/impl/delete_agent_task.py new file mode 100644 index 000000000..704180884 --- /dev/null +++ b/app/agent/tools/impl/delete_agent_task.py @@ -0,0 +1,50 @@ +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 DeleteAgentTaskInput(BaseModel): + """删除 Agent 自主定时任务的输入参数。""" + + task_id: int = Field(..., ge=1, description="ID of the task to permanently delete.") + + +class DeleteAgentTaskTool(MoviePilotTool): + """永久删除 Agent 自主定时任务。""" + + name: str = "delete_agent_task" + tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, 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." + ) + args_schema: Type[BaseModel] = DeleteAgentTaskInput + require_admin: bool = True + + def get_tool_message(self, **kwargs: object) -> Optional[str]: + """生成删除定时任务的提示消息。""" + return f"删除自主定时任务：{kwargs.get('task_id', '')}" + + def _delete_task(self, task_id: int) -> bool: + """删除当前用户的任务并移除运行时调度。""" + from app.scheduler import Scheduler + + deleted = AgentTaskOper().delete( + task_id=task_id, + user_id=str(self._user_id), + ) + if deleted: + Scheduler().remove_agent_task_job(task_id) + return deleted + + async def run(self, task_id: int, **kwargs: object) -> str: + """删除 Agent 自主定时任务。""" + payload = DeleteAgentTaskInput(task_id=task_id) + deleted = await self.run_blocking("db", self._delete_task, payload.task_id) + if not deleted: + return f"Agent 定时任务 {task_id} 不存在或不属于当前用户" + return f"Agent 定时任务 {task_id} 已删除" diff --git a/app/agent/tools/impl/query_agent_tasks.py b/app/agent/tools/impl/query_agent_tasks.py new file mode 100644 index 000000000..9674b0c9b --- /dev/null +++ b/app/agent/tools/impl/query_agent_tasks.py @@ -0,0 +1,86 @@ +import json +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.core.config import settings +from app.db.agenttask_oper import AgentTaskOper + + +class QueryAgentTasksInput(BaseModel): + """查询 Agent 自主定时任务的输入参数。""" + + task_id: Optional[int] = Field( + None, + ge=1, + description="Optional task ID. Omit it to list tasks owned by the current user.", + ) + enabled: Optional[bool] = Field( + None, + description="Optional enabled-state filter used when listing tasks.", + ) + + +class QueryAgentTasksTool(MoviePilotTool): + """查询当前用户创建的 Agent 自主定时任务。""" + + name: str = "query_agent_tasks" + tags: list[str] = [ToolTag.Read, ToolTag.Scheduler, 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." + ) + args_schema: Type[BaseModel] = QueryAgentTasksInput + require_admin: bool = True + + def get_tool_message(self, **kwargs: object) -> Optional[str]: + """生成查询定时任务的提示消息。""" + task_id = kwargs.get("task_id") + return f"查询自主定时任务：{task_id}" if task_id else "查询自主定时任务" + + def _query_tasks( + self, + task_id: Optional[int], + enabled: Optional[bool], + ) -> list[dict]: + """读取当前用户的任务及运行时下一次触发时间。""" + from app.scheduler import Scheduler + + oper = AgentTaskOper() + if task_id: + task = oper.get(task_id=task_id, user_id=str(self._user_id)) + tasks = [task] if task else [] + else: + tasks = oper.list(user_id=str(self._user_id), enabled=enabled) + scheduler = Scheduler() + result = [] + for task in tasks: + data = oper.to_dict( + task, + next_run_at=scheduler.get_agent_task_next_run(task.id), + timezone=settings.TZ, + ) + result.append(data) + return result + + async def run( + self, + task_id: Optional[int] = None, + enabled: Optional[bool] = None, + **kwargs: object, + ) -> str: + """查询 Agent 自主定时任务。""" + payload = QueryAgentTasksInput(task_id=task_id, enabled=enabled) + tasks = await self.run_blocking( + "db", + self._query_tasks, + payload.task_id, + payload.enabled, + ) + return json.dumps( + {"total": len(tasks), "tasks": tasks}, + ensure_ascii=False, + indent=2, + ) diff --git a/app/agent/tools/impl/update_agent_task.py b/app/agent/tools/impl/update_agent_task.py new file mode 100644 index 000000000..da44bfa67 --- /dev/null +++ b/app/agent/tools/impl/update_agent_task.py @@ -0,0 +1,188 @@ +import json +from datetime import datetime, timedelta +from typing import Literal, Optional, Type + +import pytz +from pydantic import BaseModel, Field, model_validator + +from app.agent.tools.base import MoviePilotTool +from app.agent.tools.tags import ToolTag +from app.core.config import settings +from app.db.agenttask_oper import AgentTaskOper +from app.utils.timer import TimerUtils + + +class UpdateAgentTaskInput(BaseModel): + """更新 Agent 自主定时任务的输入参数。""" + + task_id: int = Field(..., ge=1, description="ID of the task to update.") + name: Optional[str] = Field(None, min_length=1, max_length=100) + content: Optional[str] = Field(None, min_length=1, max_length=10000) + trigger_type: Optional[Literal["date", "cron"]] = Field( + None, + description="New trigger type. Must be provided together with trigger.", + ) + trigger: Optional[str] = Field( + None, + min_length=1, + max_length=200, + description="New ISO 8601 date or five-field cron expression.", + ) + delay_minutes: Optional[int] = Field( + None, + ge=1, + le=525600, + description=( + "For a one-time date task expressed as 'in N minutes', provide this instead " + "of trigger together with trigger_type='date'." + ), + ) + enabled: Optional[bool] = Field( + None, + description="Set false to pause the task or true to resume it.", + ) + + @model_validator(mode="after") + def validate_update(self) -> "UpdateAgentTaskInput": + """校验更新内容和触发参数组合。""" + if self.name is not None: + self.name = self.name.strip() + if not self.name: + raise ValueError("name 不能只包含空白字符") + if self.content is not None: + self.content = self.content.strip() + if not self.content: + raise ValueError("content 不能只包含空白字符") + has_schedule_update = any( + value is not None + for value in (self.trigger_type, self.trigger, self.delay_minutes) + ) + if has_schedule_update: + if self.trigger_type is None: + raise ValueError("修改触发配置时必须提供 trigger_type") + if self.trigger_type == "date": + if (self.trigger is None) == (self.delay_minutes is None): + raise ValueError("date 任务必须且只能提供 trigger 或 delay_minutes 之一") + if self.delay_minutes is not None: + timezone = pytz.timezone(settings.TZ) + self.trigger = ( + datetime.now(timezone) + timedelta(minutes=self.delay_minutes) + ).isoformat(timespec="seconds") + elif self.trigger is None or self.delay_minutes is not None: + raise ValueError("cron 任务必须提供 trigger，且不能提供 delay_minutes") + if all( + value is None + for value in ( + self.name, + self.content, + self.trigger_type, + self.enabled, + ) + ): + raise ValueError("至少需要提供一个要更新的字段") + return self + + +class UpdateAgentTaskTool(MoviePilotTool): + """修改、暂停或恢复 Agent 自主定时任务。""" + + name: str = "update_agent_task" + tags: list[str] = [ToolTag.Write, ToolTag.Scheduler, 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 " + "and enabled=true to resume." + ) + args_schema: Type[BaseModel] = UpdateAgentTaskInput + require_admin: bool = True + + def get_tool_message(self, **kwargs: object) -> Optional[str]: + """生成更新定时任务的提示消息。""" + return f"更新自主定时任务：{kwargs.get('task_id', '')}" + + def _update_task(self, payload: UpdateAgentTaskInput) -> Optional[dict]: + """更新当前用户的任务并刷新运行时调度。""" + from app.scheduler import Scheduler + + oper = AgentTaskOper() + task = oper.get(task_id=payload.task_id, user_id=str(self._user_id)) + if not task: + return None + if task.last_status == "running": + return {"error": f"Agent 定时任务 {payload.task_id} 正在执行，请稍后再修改"} + + trigger_type = payload.trigger_type or task.trigger_type + trigger_value = payload.trigger or ( + task.cron_expression if trigger_type == "cron" else task.run_at + ) + enabled = task.enabled if payload.enabled is None else payload.enabled + normalized_type, normalized_trigger = TimerUtils.normalize_schedule_trigger( + trigger_type=trigger_type, + trigger_value=trigger_value, + timezone_name=settings.TZ, + require_future=bool(enabled and trigger_type == "date"), + ) + + update_payload = {} + if payload.name is not None: + update_payload["name"] = payload.name.strip() + if payload.content is not None: + update_payload["content"] = payload.content.strip() + if payload.trigger is not None: + update_payload.update( + { + "trigger_type": normalized_type, + "cron_expression": ( + normalized_trigger if normalized_type == "cron" else None + ), + "run_at": normalized_trigger if normalized_type == "date" else None, + "last_status": "waiting", + "last_result": None, + } + ) + if payload.enabled is not None: + update_payload["enabled"] = payload.enabled + if payload.enabled: + update_payload["last_status"] = "waiting" + + oper.update( + task_id=payload.task_id, + payload=update_payload, + user_id=str(self._user_id), + ) + scheduler = Scheduler() + next_run_at = scheduler.update_agent_task_job(payload.task_id) + updated_task = oper.get(task_id=payload.task_id, user_id=str(self._user_id)) + return oper.to_dict( + updated_task, + next_run_at=next_run_at, + timezone=settings.TZ, + ) + + async def run( + self, + task_id: int, + name: Optional[str] = None, + content: Optional[str] = None, + trigger_type: Optional[str] = None, + trigger: Optional[str] = None, + delay_minutes: Optional[int] = None, + enabled: Optional[bool] = None, + **kwargs: object, + ) -> str: + """更新 Agent 自主定时任务。""" + payload = UpdateAgentTaskInput( + task_id=task_id, + name=name, + content=content, + trigger_type=trigger_type, + trigger=trigger, + delay_minutes=delay_minutes, + enabled=enabled, + ) + task = await self.run_blocking("db", self._update_task, payload) + if not task: + return f"Agent 定时任务 {task_id} 不存在或不属于当前用户" + if task.get("error"): + return task["error"] + return json.dumps(task, ensure_ascii=False, indent=2) diff --git a/app/db/agenttask_oper.py b/app/db/agenttask_oper.py new file mode 100644 index 000000000..8facf9184 --- /dev/null +++ b/app/db/agenttask_oper.py @@ -0,0 +1,150 @@ +from datetime import datetime +from typing import Optional + +from app.db import DbOper +from app.db.models.agenttask import AgentTask + + +class AgentTaskOper(DbOper): + """ + Agent 自主定时任务管理。 + """ + + @staticmethod + def _now() -> str: + """生成当前数据库时间字符串。""" + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + def add(self, **kwargs: object) -> AgentTask: + """ + 新增 Agent 定时任务。 + """ + now = self._now() + task_id = AgentTask.add_task( + self._db, + **kwargs, + enabled=True, + last_status="waiting", + run_count=0, + created_at=now, + updated_at=now, + ) + return self.get(task_id) + + def get( + self, + task_id: int, + user_id: Optional[str] = None, + ) -> Optional[AgentTask]: + """ + 查询单个 Agent 定时任务。 + """ + return AgentTask.get_for_user(self._db, task_id=task_id, user_id=user_id) + + def list( + self, + user_id: Optional[str] = None, + enabled: Optional[bool] = None, + ) -> list[AgentTask]: + """ + 查询 Agent 定时任务列表。 + """ + return AgentTask.list_for_user(self._db, user_id=user_id, enabled=enabled) + + def update( + self, + task_id: int, + payload: dict, + user_id: Optional[str] = None, + ) -> bool: + """ + 更新 Agent 定时任务。 + """ + normalized_payload = { + key: value + for key, value in payload.items() + if key in { + "name", + "content", + "trigger_type", + "cron_expression", + "run_at", + "enabled", + "last_status", + "last_result", + } + } + if not normalized_payload: + return False + normalized_payload["updated_at"] = self._now() + return AgentTask.update_task( + self._db, + task_id=task_id, + payload=normalized_payload, + user_id=user_id, + ) + + def delete(self, task_id: int, user_id: Optional[str] = None) -> bool: + """ + 删除 Agent 定时任务。 + """ + return AgentTask.delete_task( + self._db, + task_id=task_id, + user_id=user_id, + ) + + def mark_running(self, task_id: int) -> bool: + """ + 将 Agent 定时任务标记为运行中。 + """ + return AgentTask.mark_running( + self._db, + task_id=task_id, + run_at=self._now(), + ) + + def finish( + self, + task_id: int, + success: bool, + result: str, + disable: bool = False, + ) -> bool: + """ + 记录 Agent 定时任务执行结果。 + """ + return AgentTask.finish_task( + self._db, + task_id=task_id, + success=success, + result=(result or "")[:20000], + disable=disable, + ) + + @staticmethod + def to_dict( + task: AgentTask, + next_run_at: Optional[str] = None, + timezone: Optional[str] = None, + ) -> dict: + """ + 将 Agent 定时任务转换为工具可返回的结构。 + """ + return { + "id": task.id, + "name": task.name, + "content": task.content, + "trigger_type": task.trigger_type, + "cron_expression": task.cron_expression, + "run_at": task.run_at, + "timezone": timezone, + "enabled": bool(task.enabled), + "last_status": task.last_status, + "last_run_at": task.last_run_at, + "last_result": task.last_result, + "run_count": task.run_count or 0, + "next_run_at": next_run_at, + "created_at": task.created_at, + "updated_at": task.updated_at, + } diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index 72166dc30..6dfd8d549 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -1,4 +1,5 @@ from .agentchat import AgentChat +from .agenttask import AgentTask from .downloadfailure import DownloadFailure from .downloadhistory import DownloadHistory, DownloadFiles from .mediaserver import MediaServerItem diff --git a/app/db/models/agenttask.py b/app/db/models/agenttask.py new file mode 100644 index 000000000..c1d58159c --- /dev/null +++ b/app/db/models/agenttask.py @@ -0,0 +1,170 @@ +from datetime import datetime +from typing import Optional + +from sqlalchemy import Boolean, Column, Index, Integer, String, Text +from sqlalchemy.orm import Session + +from app.db import Base, db_query, db_update, get_id_column + + +class AgentTask(Base): + """ + Agent 自主定时任务表。 + """ + + id = get_id_column() + # 任务名称 + name = Column(String, nullable=False) + # 交给 Agent 执行的完整任务内容 + content = Column(Text, nullable=False) + # 触发类型：date-单次触发，cron-周期触发 + trigger_type = Column(String, nullable=False) + # 标准五段 cron 表达式 + cron_expression = Column(String) + # 单次触发时间，使用带时区的 ISO 8601 格式 + run_at = Column(String) + # 是否继续接受调度 + enabled = Column(Boolean, nullable=False, default=True) + # 创建任务的用户与会话上下文 + user_id = Column(String, nullable=False) + username = Column(String) + session_id = Column(String, nullable=False) + channel = Column(String) + source = Column(String) + original_chat_id = Column(String) + # 最近一次执行状态与结果 + last_status = Column(String, nullable=False, default="waiting") + last_run_at = Column(String) + last_result = Column(Text) + run_count = Column(Integer, nullable=False, default=0) + created_at = Column(String, nullable=False) + updated_at = Column(String, nullable=False) + + __table_args__ = ( + Index("ix_agenttask_enabled", "enabled"), + Index("ix_agenttask_user_created", "user_id", "created_at", "id"), + ) + + @classmethod + @db_update + def add_task(cls, db: Session, **kwargs: object) -> int: + """ + 新增 Agent 定时任务并返回任务 ID。 + """ + task = cls(**kwargs) + db.add(task) + db.flush() + return task.id + + @classmethod + @db_query + def get_for_user( + cls, + db: Session, + task_id: int, + user_id: Optional[str] = None, + ) -> Optional["AgentTask"]: + """ + 按任务 ID 和可选用户 ID 查询 Agent 定时任务。 + """ + query = db.query(cls).filter(cls.id == task_id) + if user_id is not None: + query = query.filter(cls.user_id == user_id) + return query.first() + + @classmethod + @db_query + def list_for_user( + cls, + db: Session, + user_id: Optional[str] = None, + enabled: Optional[bool] = None, + ) -> list["AgentTask"]: + """ + 按用户和启用状态查询 Agent 定时任务。 + """ + query = db.query(cls) + if user_id is not None: + query = query.filter(cls.user_id == user_id) + if enabled is not None: + query = query.filter(cls.enabled.is_(enabled)) + return query.order_by(cls.created_at.desc(), cls.id.desc()).all() + + @classmethod + @db_update + def update_task( + cls, + db: Session, + task_id: int, + payload: dict, + user_id: Optional[str] = None, + ) -> bool: + """ + 按任务 ID 和可选用户 ID 更新 Agent 定时任务。 + """ + query = db.query(cls).filter(cls.id == task_id) + if user_id is not None: + query = query.filter(cls.user_id == user_id) + return bool(query.update(payload)) + + @classmethod + @db_update + def delete_task( + cls, + db: Session, + task_id: int, + user_id: Optional[str] = None, + ) -> bool: + """ + 按任务 ID 和可选用户 ID 删除 Agent 定时任务。 + """ + query = db.query(cls).filter(cls.id == task_id) + if user_id is not None: + query = query.filter(cls.user_id == user_id) + return bool(query.delete()) + + @classmethod + @db_update + def mark_running(cls, db: Session, task_id: int, run_at: str) -> bool: + """ + 将可执行任务标记为运行中。 + """ + updated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + return bool( + db.query(cls) + .filter( + cls.id == task_id, + cls.enabled.is_(True), + cls.last_status != "running", + ) + .update( + { + "last_status": "running", + "last_run_at": run_at, + "updated_at": updated_at, + } + ) + ) + + @classmethod + @db_update + def finish_task( + cls, + db: Session, + task_id: int, + success: bool, + result: str, + disable: bool = False, + ) -> bool: + """ + 记录 Agent 定时任务执行结果，并按需关闭单次任务。 + """ + payload = { + "last_status": "success" if success else "failed", + "last_result": result, + "run_count": cls.run_count + 1, + "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + } + if disable: + payload["enabled"] = False + return bool(db.query(cls).filter(cls.id == task_id).update(payload)) diff --git a/app/scheduler.py b/app/scheduler.py index 49ac8f367..44de0ef14 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -28,6 +28,7 @@ from app.core.config import settings, global_vars from app.core.event import Event, eventmanager from app.core.plugin import PluginManager from app.db import SessionFactory +from app.db.agenttask_oper import AgentTaskOper from app.db.models.downloadhistory import DownloadHistory, DownloadFiles from app.db.models.message import Message from app.db.models.siteuserdata import SiteUserData @@ -48,6 +49,7 @@ from app.utils.timer import TimerUtils lock = threading.Lock() SCHEDULER_PROGRESS_PREFIX = "scheduler" +AGENT_TASK_JOB_PREFIX = "agent-task" class SchedulerChain(ChainBase): @@ -705,6 +707,10 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): # 初始化工作流服务 self.init_workflow_jobs() + # 恢复 Agent 自主定时任务 + if settings.AI_AGENT_ENABLE: + self.init_agent_task_jobs() + # 初始化插件服务 self.init_plugin_jobs() @@ -980,6 +986,145 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): # 运行结束 self.__finish_job(job_id=job_id, success=success, error=error) + @staticmethod + def _get_agent_task_job_id(task_id: int) -> str: + """生成 Agent 自主定时任务的调度器 Job ID。""" + return f"{AGENT_TASK_JOB_PREFIX}-{task_id}" + + def init_agent_task_jobs(self) -> None: + """ + 从数据库恢复所有启用的 Agent 自主定时任务。 + """ + oper = AgentTaskOper() + for task in oper.list(enabled=True): + if task.last_status == "running": + oper.update( + task_id=task.id, + payload={ + "last_status": "waiting", + "last_result": "服务重启后恢复调度", + }, + ) + self.update_agent_task_job(task.id) + + def update_agent_task_job(self, task_id: int) -> Optional[str]: + """ + 按数据库中的最新配置新增或替换 Agent 自主定时任务。 + + :param task_id: Agent 定时任务 ID + :return: 下一次执行时间，不可调度时返回 None + """ + self.remove_agent_task_job(task_id) + task = AgentTaskOper().get(task_id) + if ( + not settings.AI_AGENT_ENABLE + or not task + or not task.enabled + or not self._scheduler + ): + return None + + trigger_value = ( + task.cron_expression if task.trigger_type == "cron" else task.run_at + ) + try: + trigger = TimerUtils.build_schedule_trigger( + trigger_type=task.trigger_type, + trigger_value=trigger_value, + timezone_name=settings.TZ, + ) + except (TypeError, ValueError) as err: + logger.error(f"Agent 定时任务 {task_id} 的触发配置无效：{str(err)}") + return None + + job_id = self._get_agent_task_job_id(task_id) + with self._lock: + self._jobs[job_id] = { + "name": task.name, + "provider_name": "[Agent]", + "func": self.execute_agent_task, + "running": False, + "kwargs": {"task_id": task_id}, + } + self._scheduler.add_job( + self.start, + trigger=trigger, + id=job_id, + name=task.name, + kwargs={"job_id": job_id, "task_id": task_id}, + coalesce=True, + max_instances=1, + misfire_grace_time=None, + replace_existing=True, + ) + return self.get_agent_task_next_run(task_id) + + def remove_agent_task_job(self, task_id: int) -> None: + """ + 从运行时调度器移除 Agent 自主定时任务。 + + :param task_id: Agent 定时任务 ID + """ + job_id = self._get_agent_task_job_id(task_id) + with self._lock: + self._jobs.pop(job_id, None) + if not self._scheduler: + return + try: + self._scheduler.remove_job(job_id) + except JobLookupError: + pass + + def get_agent_task_next_run(self, task_id: int) -> Optional[str]: + """ + 查询 Agent 自主定时任务的下一次执行时间。 + + :param task_id: Agent 定时任务 ID + :return: 带时区的 ISO 8601 时间，不再执行时返回 None + """ + job_id = self._get_agent_task_job_id(task_id) + if self._scheduler: + job = self._scheduler.get_job(job_id) + next_run_time = getattr(job, "next_run_time", None) if job else None + if next_run_time: + return next_run_time.isoformat(timespec="seconds") + + task = AgentTaskOper().get(task_id) + if not task or not task.enabled: + return None + trigger_value = ( + task.cron_expression if task.trigger_type == "cron" else task.run_at + ) + try: + next_run_time = TimerUtils.get_schedule_next_run_time( + trigger_type=task.trigger_type, + trigger_value=trigger_value, + timezone_name=settings.TZ, + ) + except (TypeError, ValueError): + return None + return ( + next_run_time.isoformat(timespec="seconds") + if next_run_time + else None + ) + + async def execute_agent_task(self, task_id: int) -> tuple[bool, str]: + """ + 唤醒 Agent 执行指定自主定时任务。 + + :param task_id: Agent 定时任务 ID + :return: 执行是否成功及结果摘要 + """ + from app.agent import agent_manager + + try: + return await agent_manager.execute_scheduled_task(task_id) + finally: + task = AgentTaskOper().get(task_id) + if task and task.trigger_type == "date" and not task.enabled: + self.remove_agent_task_job(task_id) + def init_plugin_jobs(self): """ 初始化插件定时服务 diff --git a/app/utils/timer.py b/app/utils/timer.py index e220e3161..80f20f403 100644 --- a/app/utils/timer.py +++ b/app/utils/timer.py @@ -1,9 +1,117 @@ import datetime import random -from typing import List +from typing import List, Optional, Tuple, Union + +import pytz +from apscheduler.triggers.cron import CronTrigger +from apscheduler.triggers.date import DateTrigger class TimerUtils: + """ + 定时与时间差计算工具。 + """ + + SCHEDULE_TRIGGER_TYPES = ("date", "cron") + + @staticmethod + def normalize_schedule_trigger( + trigger_type: str, + trigger_value: str, + timezone_name: str, + require_future: bool = False, + ) -> Tuple[str, str]: + """ + 校验并规范化单次时间或五段 cron 表达式。 + + :param trigger_type: 触发类型，支持 date 或 cron + :param trigger_value: 带时区或本地时间字符串，或五段 cron 表达式 + :param timezone_name: 无显式时区时使用的系统时区 + :param require_future: 单次任务是否必须安排在未来 + :return: 规范化后的触发类型和触发值 + """ + normalized_type = str(trigger_type or "").strip().lower() + normalized_value = str(trigger_value or "").strip() + if normalized_type not in TimerUtils.SCHEDULE_TRIGGER_TYPES: + raise ValueError("trigger_type 仅支持 date 或 cron") + if not normalized_value: + raise ValueError("trigger 不能为空") + + timezone = pytz.timezone(timezone_name) + if normalized_type == "cron": + normalized_value = " ".join(normalized_value.split()) + if len(normalized_value.split()) != 5: + raise ValueError("cron 必须是标准五段表达式：分 时 日 月 周") + CronTrigger.from_crontab(normalized_value, timezone=timezone) + return normalized_type, normalized_value + + try: + run_at = datetime.datetime.fromisoformat(normalized_value) + except ValueError as err: + raise ValueError( + "date 时间必须使用 ISO 8601 格式，例如 2026-07-19 20:30:00" + ) from err + if run_at.tzinfo is None: + run_at = timezone.localize(run_at) + else: + run_at = run_at.astimezone(timezone) + if require_future and run_at <= datetime.datetime.now(timezone): + raise ValueError("单次任务的触发时间必须晚于当前时间") + return normalized_type, run_at.isoformat(timespec="seconds") + + @staticmethod + def build_schedule_trigger( + trigger_type: str, + trigger_value: str, + timezone_name: str, + ) -> Union[CronTrigger, DateTrigger]: + """ + 构建 APScheduler 单次或 cron 触发器。 + + :param trigger_type: 触发类型，支持 date 或 cron + :param trigger_value: 已配置的触发时间或 cron 表达式 + :param timezone_name: 调度器使用的系统时区 + :return: APScheduler 触发器 + """ + normalized_type, normalized_value = TimerUtils.normalize_schedule_trigger( + trigger_type=trigger_type, + trigger_value=trigger_value, + timezone_name=timezone_name, + ) + timezone = pytz.timezone(timezone_name) + if normalized_type == "cron": + return CronTrigger.from_crontab(normalized_value, timezone=timezone) + return DateTrigger( + run_date=datetime.datetime.fromisoformat(normalized_value), + timezone=timezone, + ) + + @staticmethod + def get_schedule_next_run_time( + trigger_type: str, + trigger_value: str, + timezone_name: str, + now: Optional[datetime.datetime] = None, + ) -> Optional[datetime.datetime]: + """ + 计算指定触发配置的下一次执行时间。 + + :param trigger_type: 触发类型，支持 date 或 cron + :param trigger_value: 已配置的触发时间或 cron 表达式 + :param timezone_name: 调度器使用的系统时区 + :param now: 可选的计算基准时间 + :return: 下一次执行时间，不再触发时返回 None + """ + timezone = pytz.timezone(timezone_name) + current_time = now or datetime.datetime.now(timezone) + if current_time.tzinfo is None: + current_time = timezone.localize(current_time) + trigger = TimerUtils.build_schedule_trigger( + trigger_type=trigger_type, + trigger_value=trigger_value, + timezone_name=timezone_name, + ) + return trigger.get_next_fire_time(None, current_time) @staticmethod def random_scheduler(num_executions: int = 1, diff --git a/database/versions/c4e8f7a1b2d3_2_2_12.py b/database/versions/c4e8f7a1b2d3_2_2_12.py new file mode 100644 index 000000000..e6cd2bb7f --- /dev/null +++ b/database/versions/c4e8f7a1b2d3_2_2_12.py @@ -0,0 +1,67 @@ +"""2.2.12 +新增 Agent 自主定时任务表 + +Revision ID: c4e8f7a1b2d3 +Revises: b7d4a9c2e6f1 +Create Date: 2026-07-19 +""" + +from alembic import op +import sqlalchemy as sa + +revision = "c4e8f7a1b2d3" +down_revision = "b7d4a9c2e6f1" +branch_labels = None +depends_on = None + + +def _has_table(inspector: sa.Inspector, table_name: str) -> bool: + """检查数据表是否已存在。""" + return table_name in inspector.get_table_names() + + +def upgrade() -> None: + """升级数据库结构。""" + inspector = sa.inspect(op.get_bind()) + if _has_table(inspector, "agenttask"): + return + + op.create_table( + "agenttask", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(), nullable=False), + sa.Column("content", sa.Text(), nullable=False), + sa.Column("trigger_type", sa.String(), nullable=False), + sa.Column("cron_expression", sa.String(), nullable=True), + sa.Column("run_at", sa.String(), nullable=True), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column("user_id", sa.String(), nullable=False), + sa.Column("username", sa.String(), nullable=True), + sa.Column("session_id", sa.String(), nullable=False), + sa.Column("channel", sa.String(), nullable=True), + sa.Column("source", sa.String(), nullable=True), + sa.Column("original_chat_id", sa.String(), nullable=True), + sa.Column("last_status", sa.String(), nullable=False), + sa.Column("last_run_at", sa.String(), nullable=True), + sa.Column("last_result", sa.Text(), nullable=True), + sa.Column("run_count", sa.Integer(), nullable=False), + sa.Column("created_at", sa.String(), nullable=False), + sa.Column("updated_at", sa.String(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_agenttask_enabled", "agenttask", ["enabled"]) + op.create_index( + "ix_agenttask_user_created", + "agenttask", + ["user_id", "created_at", "id"], + ) + + +def downgrade() -> None: + """回滚数据库结构。""" + inspector = sa.inspect(op.get_bind()) + if not _has_table(inspector, "agenttask"): + return + op.drop_index("ix_agenttask_user_created", table_name="agenttask") + op.drop_index("ix_agenttask_enabled", table_name="agenttask") + op.drop_table("agenttask") diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 5cad693d7..cd718231c 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -181,6 +181,35 @@ FastAPI 异常响应保留 `detail` 字段，并在错误详情为文本时返 工具的 `inputSchema` 只包含实际执行业务所需的参数，不包含用于解释调用原因的通用 `explanation` 参数，以减少 Agent 上下文消耗。 +#### Agent 自主定时任务工具 + +以下工具用于管理会在指定时间重新唤醒 Agent 的持久化任务，均为管理员级工具： + +| 工具 | 说明 | +| :--- | :--- | +| `create_agent_task` | 创建单次或周期任务，并保存任务内容及当前用户、会话和消息渠道 | +| `query_agent_tasks` | 查询任务配置、启用状态、下次执行时间及最近执行结果 | +| `update_agent_task` | 修改任务内容或触发器，也可通过 `enabled` 暂停、恢复任务 | +| `delete_agent_task` | 永久删除任务并立即移除运行时调度 | + +`trigger_type=date` 表示单次执行：“30 分钟后检查”这类相对时间传 `delay_minutes=30`，由后端计算精确时间；固定时间则传 ISO 8601 `trigger`，支持精确到秒。`trigger_type=cron` 使用标准五段 cron（分、时、日、月、周），适合周期检查。未显式携带时区的时间按 MoviePilot 的 `TZ` 配置解释。任务由内存调度器精确触发，配置持久化到数据库，服务重启后会自动恢复；触发后 Agent 在原会话中执行 `content` 并把结果发送到原消息渠道。通过无会话上下文的 MCP/CLI 创建时，结果改发到 MoviePilot 已配置的管理员通知渠道。 + +创建单次任务的参数示例： + +```json +{ + "tool_name": "create_agent_task", + "arguments": { + "name": "检查电影资源", + "content": "搜索电影《示例电影》是否已有可下载资源，并报告站点、版本和大小；不要自动下载。", + "trigger_type": "date", + "delay_minutes": 30 + } +} +``` + +创建每天 20:30 执行的周期任务时，使用 `trigger_type=cron` 和 `trigger="30 20 * * *"`。 + **认证**: 需要API KEY，在请求头中添加 `X-API-KEY: ` 或在查询参数中添加 `apikey=` **响应示例**: diff --git a/skills/moviepilot-cli/SKILL.md b/skills/moviepilot-cli/SKILL.md index 8a2ffe1d6..b4faf42e9 100644 --- a/skills/moviepilot-cli/SKILL.md +++ b/skills/moviepilot-cli/SKILL.md @@ -1,6 +1,6 @@ --- name: moviepilot-cli -version: 3 +version: 4 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, 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, delete_agent_task, query_workflows, run_workflow, query_rule_groups, query_episode_schedule, send_message | ## Workflows @@ -186,6 +186,32 @@ Trigger a search for missing episodes (confirm with user first): Remove a subscription (confirm with user first): `moviepilot tool run delete_subscribe subscribe_id=123` +### Manage Autonomous Agent Tasks + +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. + +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` + +For a one-time task at a fixed time, use `date` with an ISO 8601 `trigger`: +`moviepilot tool run create_agent_task name="今晚检查资源" content="检查目标电影是否有资源并报告。" trigger_type=date trigger="2026-07-19 20:30:00"` + +For recurring work, use a standard five-field cron expression. This example +runs every day at 20:30: +`moviepilot tool run create_agent_task name="每日资源检查" content="检查目标电影是否有资源并报告。" trigger_type=cron trigger="30 20 * * *"` + +List tasks and inspect `next_run_at` and the latest result: +`moviepilot tool run query_agent_tasks` + +Pause or resume a task: +`moviepilot tool run update_agent_task task_id=1 enabled=false` + +Delete a task only after confirming permanent removal with the user: +`moviepilot tool run delete_agent_task task_id=1` + ### Check Library and Subscriptions Run before any download or subscription to avoid duplicates. diff --git a/tests/test_agent_scheduled_tasks.py b/tests/test_agent_scheduled_tasks.py new file mode 100644 index 000000000..45a8e1818 --- /dev/null +++ b/tests/test_agent_scheduled_tasks.py @@ -0,0 +1,340 @@ +import json +import threading +from datetime import datetime, timedelta +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +import pytz +from apscheduler.schedulers.background import BackgroundScheduler + +from app.agent import AgentChain, AgentManager, ReplyMode +from app.agent.tools.factory import MoviePilotToolFactory +from app.agent.tools.impl.create_agent_task import ( + CreateAgentTaskInput, + 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.update_agent_task import UpdateAgentTaskTool +from app.core.config import settings +from app.db.agenttask_oper import AgentTaskOper +from app.scheduler import Scheduler +from app.utils.timer import TimerUtils + + +class _FakeAgentTaskScheduler: + """记录 Agent 定时任务工具触发的运行时调度变更。""" + + def __init__(self) -> None: + """初始化运行时调度记录。""" + self.updated = [] + self.removed = [] + + def update_agent_task_job(self, task_id: int) -> str: + """记录任务重载并返回固定的下一次执行时间。""" + self.updated.append(task_id) + return "2099-01-01T00:00:00+08:00" + + def remove_agent_task_job(self, task_id: int) -> None: + """记录任务移除。""" + self.removed.append(task_id) + + def get_agent_task_next_run(self, task_id: int) -> str: + """返回固定的下一次执行时间。""" + return "2099-01-01T00:00:00+08:00" + + +@pytest.fixture +def anyio_backend() -> str: + """限定异步用例使用项目 Agent 运行时采用的 asyncio 后端。""" + return "asyncio" + + +@pytest.fixture(autouse=True) +def enable_ai_agent(monkeypatch) -> None: + """在当前测试模块中启用 Agent 调度能力并在用例后自动还原。""" + monkeypatch.setattr(settings, "AI_AGENT_ENABLE", True) + + +def _future_time(minutes: int = 10) -> str: + """生成系统时区内的未来时间字符串。""" + timezone = pytz.timezone(settings.TZ) + return (datetime.now(timezone) + timedelta(minutes=minutes)).isoformat( + timespec="seconds" + ) + + +def _build_tool(tool_class, user_id: str): + """构造带当前用户消息上下文的 Agent 工具。""" + tool = tool_class(session_id=f"session-{user_id}", user_id=user_id) + tool.set_message_attr( + channel="Telegram", + source="telegram-test", + username="admin", + ) + tool.set_agent_context({"is_admin": True}) + return tool + + +def test_timer_utils_validates_date_and_cron_triggers() -> None: + """自主任务时间工具应规范化单次时间并校验五段 cron。""" + trigger_type, trigger = TimerUtils.normalize_schedule_trigger( + trigger_type="date", + trigger_value=_future_time(), + timezone_name=settings.TZ, + require_future=True, + ) + assert trigger_type == "date" + assert datetime.fromisoformat(trigger).tzinfo is not None + + trigger_type, trigger = TimerUtils.normalize_schedule_trigger( + trigger_type="cron", + trigger_value=" 30 20 * * * ", + timezone_name=settings.TZ, + ) + assert trigger_type == "cron" + assert trigger == "30 20 * * *" + + with pytest.raises(ValueError, match="标准五段"): + TimerUtils.normalize_schedule_trigger( + trigger_type="cron", + trigger_value="30 20 * *", + timezone_name=settings.TZ, + ) + + +def test_agent_task_tools_are_registered_with_relative_delay_schema() -> None: + """工具工厂应公开完整任务管理工具，并声明相对分钟参数。""" + tool_names = { + tool_class.model_fields["name"].default + for tool_class in MoviePilotToolFactory.BUILTIN_TOOL_CLASSES + } + assert { + "create_agent_task", + "query_agent_tasks", + "update_agent_task", + "delete_agent_task", + }.issubset(tool_names) + assert "delay_minutes" in CreateAgentTaskInput.model_json_schema()["properties"] + + +def test_agent_task_oper_persists_and_scopes_tasks() -> None: + """AgentTaskOper 应持久化任务并按创建用户隔离查询和修改。""" + user_id = f"user-{uuid4().hex}" + oper = AgentTaskOper() + task = oper.add( + name="检查资源", + content="检查电影资源", + trigger_type="cron", + cron_expression="0 */2 * * *", + run_at=None, + user_id=user_id, + username="admin", + session_id=f"session-{user_id}", + channel="Telegram", + source="telegram-test", + original_chat_id="chat-1", + ) + + assert oper.get(task.id, user_id=user_id).content == "检查电影资源" + assert oper.get(task.id, user_id="another-user") is None + assert [item.id for item in oper.list(user_id=user_id)] == [task.id] + + assert oper.update( + task_id=task.id, + user_id=user_id, + payload={"content": "检查更新后的资源", "unknown": "ignored"}, + ) + assert oper.get(task.id).content == "检查更新后的资源" + assert oper.mark_running(task.id) + assert not oper.mark_running(task.id) + assert oper.finish(task.id, success=True, result="完成") + assert not oper.delete(task.id, user_id="another-user") + assert oper.delete(task.id, user_id=user_id) + + +def test_scheduler_registers_and_removes_agent_task_job() -> None: + """Scheduler 应把数据库任务注册为精确 APScheduler Job 并可动态移除。""" + user_id = f"user-{uuid4().hex}" + task = AgentTaskOper().add( + name="定时检查", + content="检查资源", + trigger_type="date", + cron_expression=None, + run_at=_future_time(), + user_id=user_id, + username="admin", + session_id=f"session-{user_id}", + channel="Telegram", + source="telegram-test", + original_chat_id="chat-1", + ) + scheduler = object.__new__(Scheduler) + scheduler._lock = threading.RLock() + scheduler._jobs = {} + scheduler._scheduler = BackgroundScheduler(timezone=settings.TZ) + + next_run_at = scheduler.update_agent_task_job(task.id) + job_id = scheduler._get_agent_task_job_id(task.id) + + 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.remove_agent_task_job(task.id) + assert scheduler._scheduler.get_job(job_id) is None + assert job_id not in scheduler._jobs + + +@pytest.mark.anyio +async def test_agent_task_tools_manage_persistent_schedule(monkeypatch) -> None: + """Agent 管理工具应完成创建、查询、暂停、修改和删除闭环。""" + user_id = f"user-{uuid4().hex}" + fake_scheduler = _FakeAgentTaskScheduler() + monkeypatch.setattr("app.scheduler.Scheduler", lambda: fake_scheduler) + + create_tool = _build_tool(CreateAgentTaskTool, user_id) + created = json.loads( + await create_tool.run( + name="十分钟后检查", + content="检查示例电影是否有资源，不要自动下载", + trigger_type="date", + delay_minutes=10, + ) + ) + task_id = created["id"] + assert created["enabled"] is True + assert datetime.fromisoformat(created["run_at"]) > datetime.now( + pytz.timezone(settings.TZ) + ) + assert created["next_run_at"] == "2099-01-01T00:00:00+08:00" + assert fake_scheduler.updated == [task_id] + + query_tool = _build_tool(QueryAgentTasksTool, user_id) + queried = json.loads(await query_tool.run(task_id=task_id)) + assert queried["total"] == 1 + assert queried["tasks"][0]["content"].startswith("检查示例电影") + + update_tool = _build_tool(UpdateAgentTaskTool, user_id) + updated = json.loads( + await update_tool.run( + task_id=task_id, + content="检查示例电影是否有 4K 资源，不要自动下载", + trigger_type="cron", + trigger="*/15 * * * *", + enabled=True, + ) + ) + assert updated["trigger_type"] == "cron" + assert updated["cron_expression"] == "*/15 * * * *" + assert updated["run_at"] is None + + assert AgentTaskOper().mark_running(task_id) + running_update = await update_tool.run(task_id=task_id, enabled=False) + assert running_update == f"Agent 定时任务 {task_id} 正在执行，请稍后再修改" + AgentTaskOper().finish(task_id, success=True, result="完成") + + 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 会话和渠道，并在单次执行后停用任务。""" + user_id = f"user-{uuid4().hex}" + task = AgentTaskOper().add( + name="检查电影资源", + content="搜索示例电影是否已有资源", + trigger_type="date", + cron_expression=None, + run_at=_future_time(), + 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="已找到 2 个资源") + + success, result = await manager.execute_scheduled_task(task.id) + + assert success is True + assert result == "已找到 2 个资源" + 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["reply_mode"] == ReplyMode.DISPATCH + assert kwargs["wait_for_completion"] is True + assert "搜索示例电影是否已有资源" in kwargs["message"] + + completed = AgentTaskOper().get(task.id) + assert completed.enabled is False + assert completed.last_status == "success" + assert completed.last_result == "已找到 2 个资源" + assert completed.run_count == 1 + + recurring_task = AgentTaskOper().add( + name="周期检查电影资源", + content="继续检查示例电影是否已有资源", + trigger_type="cron", + cron_expression="*/30 * * * *", + 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", + ) + + success, _ = await manager.execute_scheduled_task(recurring_task.id) + + assert success is True + recurring_completed = AgentTaskOper().get(recurring_task.id) + assert recurring_completed.enabled is True + assert recurring_completed.last_status == "success" + assert recurring_completed.run_count == 1 + + +@pytest.mark.anyio +async def test_agent_manager_dispatches_contextless_result_to_admin( + monkeypatch, +) -> None: + """无原消息渠道的 MCP 任务应捕获结果并发送到管理员通知渠道。""" + user_id = f"api-{uuid4().hex}" + task = AgentTaskOper().add( + name="后台检查资源", + content="检查示例电影是否已有资源", + trigger_type="cron", + cron_expression="0 * * * *", + run_at=None, + user_id=user_id, + username="API Client", + session_id=f"session-{user_id}", + channel=None, + source="api", + original_chat_id=None, + ) + 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 True + assert result == "后台检查完成" + kwargs = manager.process_message.await_args.kwargs + assert kwargs["reply_mode"] == ReplyMode.CAPTURE_ONLY + assert kwargs["allow_message_tools"] is False + notification = post_message.await_args.args[0] + assert notification.userid is None + assert notification.username == settings.SUPERUSER + assert notification.text == "后台检查完成" diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index d21352ea1..3945e14c6 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": "3", + "moviepilot-cli": "4", "moviepilot-update": "3", }