diff --git a/app/agent/__init__.py b/app/agent/__init__.py index d74a0bc49..717565f92 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -2941,11 +2941,16 @@ 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]: + async def execute_scheduled_task( + self, + task_id: int, + trigger_source: str = "scheduled", + ) -> tuple[bool, str]: """ 按持久化上下文唤醒 Agent 执行自主定时任务并向用户回传结果。 :param task_id: Agent 定时任务 ID + :param trigger_source: 触发入口，scheduled-自动调度，manual-显式立即执行 :return: 执行是否成功及结果摘要 """ if not settings.AI_AGENT_ENABLE: @@ -2954,23 +2959,27 @@ class AgentManager: task = oper.get(task_id) if not task or not task.enabled: return False, "Agent 定时任务不存在或已停用" - if not oper.mark_running(task_id): + run = oper.begin_run(task_id=task_id, trigger_source=trigger_source) + if not run: return False, "Agent 定时任务当前不可执行" + trigger_description = ( + "已手动触发" if run.trigger_source == "manual" else "已按计划触发" + ) task_message = ( - f"定时任务已按计划触发。请立即完成下面的任务，不要只确认收到，" + f"定时任务{trigger_description}。请立即完成下面的任务，不要只确认收到，" f"也不要重复创建同一个定时任务。\n\n" - f"任务名称：{task.name}\n" - f"任务内容：{task.content}\n\n" + f"任务名称：{run.name}\n" + f"任务内容：{run.content}\n\n" "完成后请直接向用户发送消息报告本次执行结果；如果无法完成，也需发送消息说明原因。" ) success = True result = "" - notification_username = task.username or settings.SUPERUSER + notification_username = run.username or settings.SUPERUSER try: result = await self.process_message( - session_id=task.session_id, - user_id=task.user_id, + session_id=run.session_id, + user_id=run.user_id, message=task_message, channel=None, source=None, @@ -2996,23 +3005,17 @@ class AgentManager: Notification( mtype=NotificationType.Agent, username=notification_username, - title=f"定时任务执行失败：{task.name}", + title=f"定时任务执行失败：{run.name}", text=result, save_history=False, ) ) finally: - current_task = oper.get(task_id) - oper.finish( - task_id=task_id, + oper.finish_run( + run_id=run.run_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 - ), + disable_date_task=run.trigger_type == "date", ) return success, str(result or "任务执行完成") diff --git a/app/agent/tools/impl/query_agent_tasks.py b/app/agent/tools/impl/query_agent_tasks.py index 16e88c9a3..7b4c216b6 100644 --- a/app/agent/tools/impl/query_agent_tasks.py +++ b/app/agent/tools/impl/query_agent_tasks.py @@ -64,6 +64,15 @@ class QueryAgentTasksTool(MoviePilotTool): next_run_at=scheduler.get_agent_task_next_run(task.id), timezone=settings.TZ, ) + if task_id: + data["recent_runs"] = [ + oper.run_to_dict(run) + for run in oper.list_runs( + task_id=task.id, + user_id=str(self._user_id), + limit=10, + ) + ] result.append(data) return result diff --git a/app/agent/tools/impl/update_agent_task.py b/app/agent/tools/impl/update_agent_task.py index dd111cc6c..7bcaf5906 100644 --- a/app/agent/tools/impl/update_agent_task.py +++ b/app/agent/tools/impl/update_agent_task.py @@ -164,11 +164,16 @@ class UpdateAgentTaskTool(MoviePilotTool): if payload.enabled and task.last_status != "interrupted": update_payload["last_status"] = "waiting" - oper.update( + updated = oper.update( task_id=payload.task_id, payload=update_payload, user_id=str(self._user_id), ) + if not updated: + current = oper.get(task_id=payload.task_id, user_id=str(self._user_id)) + if current and current.last_status == "running": + return {"error": f"Agent 定时任务 {payload.task_id} 正在执行，请稍后再修改"} + return None 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)) diff --git a/app/db/agenttask_oper.py b/app/db/agenttask_oper.py index 1d83939c8..512d36f0d 100644 --- a/app/db/agenttask_oper.py +++ b/app/db/agenttask_oper.py @@ -1,8 +1,12 @@ +from __future__ import annotations + from datetime import datetime from typing import Optional +from uuid import uuid4 from app.db import DbOper from app.db.models.agenttask import AgentTask +from app.db.models.agenttaskrun import AgentTaskRun class AgentTaskOper(DbOper): @@ -86,32 +90,80 @@ class AgentTaskOper(DbOper): def delete(self, task_id: int, user_id: Optional[str] = None) -> bool: """ - 删除 Agent 定时任务。 + 删除非运行中的 Agent 定时任务及其运行历史。 """ - return AgentTask.delete_task( + return AgentTaskRun.delete_task_and_runs( self._db, task_id=task_id, user_id=user_id, ) - def mark_running(self, task_id: int) -> bool: + def begin_run( + self, + task_id: int, + trigger_source: str = "scheduled", + ) -> Optional[AgentTaskRun]: """ - 将 Agent 定时任务标记为运行中。 + 原子创建一次运行并返回其任务快照。 """ - return AgentTask.mark_running( + run_id = uuid4().hex + created_run_id = AgentTaskRun.begin_run( self._db, task_id=task_id, - run_at=self._now(), + run_id=run_id, + trigger_source=trigger_source, + started_at=self._now(), ) + return self.get_run(created_run_id) if created_run_id else None + + def mark_running(self, task_id: int) -> bool: + """兼容既有调用并为该次执行创建运行记录。""" + return self.begin_run(task_id=task_id) is not None def mark_interrupted(self, task_id: int, result: str) -> bool: """ 将遗留的运行中任务标记为中断且结果未知。 """ - return AgentTask.mark_interrupted( + return AgentTaskRun.interrupt_task( self._db, task_id=task_id, result=(result or "")[:20000], + finished_at=self._now(), + ) + + def get_run(self, run_id: str) -> Optional[AgentTaskRun]: + """查询一次 Agent 任务运行。""" + return AgentTaskRun.get_by_run_id(self._db, run_id=run_id) + + def list_runs( + self, + task_id: int, + user_id: Optional[str] = None, + limit: int = 10, + ) -> list[AgentTaskRun]: + """查询任务最近的有界运行历史。""" + return AgentTaskRun.list_for_task( + self._db, + task_id=task_id, + user_id=user_id, + limit=limit, + ) + + def finish_run( + self, + run_id: str, + success: bool, + result: str, + disable_date_task: bool = False, + ) -> bool: + """收口精确运行并更新仍匹配的任务投影。""" + return AgentTaskRun.finish_run( + self._db, + run_id=run_id, + success=success, + result=(result or "")[:20000], + finished_at=self._now(), + disable_date_task=disable_date_task, ) def finish( @@ -124,12 +176,14 @@ class AgentTaskOper(DbOper): """ 记录 Agent 定时任务执行结果。 """ - return AgentTask.finish_task( - self._db, - task_id=task_id, + task = self.get(task_id) + if not task or not task.last_run_id: + return False + return self.finish_run( + run_id=task.last_run_id, success=success, - result=(result or "")[:20000], - disable=disable, + result=result, + disable_date_task=disable, ) @staticmethod @@ -153,8 +207,27 @@ class AgentTaskOper(DbOper): "last_status": task.last_status, "last_run_at": task.last_run_at, "last_result": task.last_result, + "last_run_id": task.last_run_id, "run_count": task.run_count or 0, "next_run_at": next_run_at, "created_at": task.created_at, "updated_at": task.updated_at, } + + @staticmethod + def run_to_dict(run: AgentTaskRun) -> dict: + """将一次 Agent 任务运行转换为工具返回结构。""" + return { + "run_id": run.run_id, + "task_id": run.task_id, + "trigger_source": run.trigger_source, + "name": run.name, + "content": run.content, + "trigger_type": run.trigger_type, + "cron_expression": run.cron_expression, + "run_at": run.run_at, + "status": run.status, + "started_at": run.started_at, + "finished_at": run.finished_at, + "result": run.result, + } diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index be141de0d..c882d27a2 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -1,5 +1,6 @@ from .agentchat import AgentChat from .agenttask import AgentTask +from .agenttaskrun import AgentTaskRun 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 index a47b2de79..e95ad81b3 100644 --- a/app/db/models/agenttask.py +++ b/app/db/models/agenttask.py @@ -1,4 +1,3 @@ -from datetime import datetime from typing import Optional from sqlalchemy import Boolean, Column, Index, Integer, String, Text @@ -36,6 +35,8 @@ class AgentTask(Base): last_status = Column(String, nullable=False, default="waiting") last_run_at = Column(String) last_result = Column(Text) + # 最新一次真实执行的公开 ID，用于保护 last_* 投影不被旧运行覆盖 + last_run_id = Column(String) # 已收口执行次数；进程中断的未完成尝试不计入 run_count = Column(Integer, nullable=False, default=0) created_at = Column(String, nullable=False) @@ -101,95 +102,15 @@ class AgentTask(Base): user_id: Optional[str] = None, ) -> bool: """ - 按任务 ID 和可选用户 ID 更新 Agent 定时任务。 + 仅在任务未运行时按任务 ID 和可选用户 ID 更新配置。 + + 运行状态与配置必须在同一条条件更新中判定，避免执行认领后被并发配置写入 + 覆盖回可再次执行的状态。 """ - query = db.query(cls).filter(cls.id == task_id) + query = db.query(cls).filter( + cls.id == task_id, + cls.last_status != "running", + ) 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 mark_interrupted(cls, db: Session, task_id: int, result: str) -> bool: - """ - 将服务重启时遗留的运行中任务标记为结果未知。 - - 该状态保留原执行时间和计数，避免把可能已经产生副作用的执行误记为 - 从未开始或完整失败。 - """ - return bool( - db.query(cls) - .filter( - cls.id == task_id, - cls.last_status == "running", - ) - .update( - { - "last_status": "interrupted", - "last_result": result, - "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - } - ) - ) - - @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/db/models/agenttaskrun.py b/app/db/models/agenttaskrun.py new file mode 100644 index 000000000..4b72fd95d --- /dev/null +++ b/app/db/models/agenttaskrun.py @@ -0,0 +1,254 @@ +from typing import Optional + +from sqlalchemy import Column, Index, Integer, String, Text, update +from sqlalchemy.orm import Session + +from app.db import Base, db_query, db_update, get_id_column +from app.db.models.agenttask import AgentTask + + +class AgentTaskRun(Base): + """Agent 自主定时任务的一次真实执行记录。""" + + id = get_id_column() + # 对外稳定的运行身份；内部自增主键不进入 Agent 合同 + run_id = Column(String, nullable=False) + # 所属计划及触发入口 + task_id = Column(Integer, nullable=False) + trigger_source = Column(String, nullable=False) + # 执行开始时的任务与用户上下文快照 + name = Column(String, nullable=False) + content = Column(Text, nullable=False) + trigger_type = Column(String, nullable=False) + cron_expression = Column(String) + run_at = Column(String) + user_id = Column(String, nullable=False) + username = Column(String) + session_id = Column(String, nullable=False) + channel = Column(String) + message_source = Column(String) + original_chat_id = Column(String) + # running-success/failed/interrupted；取消沿用 failed 和明确结果文本 + status = Column(String, nullable=False) + started_at = Column(String, nullable=False) + finished_at = Column(String) + result = Column(Text) + + __table_args__ = ( + Index("ix_agenttaskrun_run_id", "run_id", unique=True), + Index("ix_agenttaskrun_task_started", "task_id", "started_at", "id"), + ) + + @classmethod + @db_update + def begin_run( + cls, + db: Session, + task_id: int, + run_id: str, + trigger_source: str, + started_at: str, + ) -> Optional[str]: + """原子认领可执行任务并创建对应的运行记录。""" + if trigger_source not in {"scheduled", "manual"}: + raise ValueError(f"不支持的 Agent 任务触发来源：{trigger_source}") + # 认领和快照读取必须是同一条语句，配置更新与执行开始才能共享同一行级顺序。 + claimed = db.execute( + update(AgentTask) + .where( + AgentTask.id == task_id, + AgentTask.enabled.is_(True), + AgentTask.last_status != "running", + ) + .values({ + "last_status": "running", + "last_run_at": started_at, + "last_run_id": run_id, + "updated_at": started_at, + }) + .returning( + AgentTask.id, + AgentTask.name, + AgentTask.content, + AgentTask.trigger_type, + AgentTask.cron_expression, + AgentTask.run_at, + AgentTask.user_id, + AgentTask.username, + AgentTask.session_id, + AgentTask.channel, + AgentTask.source, + AgentTask.original_chat_id, + ) + .execution_options(synchronize_session=False) + ).mappings().first() + if not claimed: + return None + db.add(cls( + run_id=run_id, + task_id=claimed["id"], + trigger_source=trigger_source, + name=claimed["name"], + content=claimed["content"], + trigger_type=claimed["trigger_type"], + cron_expression=claimed["cron_expression"], + run_at=claimed["run_at"], + user_id=claimed["user_id"], + username=claimed["username"], + session_id=claimed["session_id"], + channel=claimed["channel"], + message_source=claimed["source"], + original_chat_id=claimed["original_chat_id"], + status="running", + started_at=started_at, + )) + db.flush() + return run_id + + @classmethod + @db_update + def finish_run( + cls, + db: Session, + run_id: str, + success: bool, + result: str, + finished_at: str, + disable_date_task: bool = False, + ) -> bool: + """原子收口精确运行，并仅在仍为最新运行时更新任务投影。""" + run = db.query(cls).filter( + cls.run_id == run_id, + ).first() + if not run: + return False + status = "success" if success else "failed" + finalized = db.query(cls).filter( + cls.run_id == run_id, + cls.status == "running", + ).update( + { + "status": status, + "result": result, + "finished_at": finished_at, + }, + synchronize_session=False, + ) + if not finalized: + return False + + task = db.query(AgentTask).filter( + AgentTask.id == run.task_id, + AgentTask.last_run_id == run_id, + ).first() + if task: + payload = { + "last_status": status, + "last_result": result, + "run_count": AgentTask.run_count + 1, + "updated_at": finished_at, + } + if ( + disable_date_task + and run.trigger_type == "date" + and task.trigger_type == run.trigger_type + and task.run_at == run.run_at + ): + payload["enabled"] = False + db.query(AgentTask).filter( + AgentTask.id == run.task_id, + AgentTask.last_run_id == run_id, + ).update(payload, synchronize_session=False) + return True + + @classmethod + @db_update + def interrupt_task( + cls, + db: Session, + task_id: int, + result: str, + finished_at: str, + ) -> bool: + """原子标记冷启动时遗留的最新运行及任务投影为结果未知。""" + task = db.query(AgentTask).filter( + AgentTask.id == task_id, + AgentTask.last_status == "running", + ).first() + if not task: + return False + if task.last_run_id: + interrupted = db.query(cls).filter( + cls.run_id == task.last_run_id, + cls.task_id == task.id, + cls.status == "running", + ).update( + { + "status": "interrupted", + "result": result, + "finished_at": finished_at, + }, + synchronize_session=False, + ) + if not interrupted: + return False + return bool(db.query(AgentTask).filter( + AgentTask.id == task.id, + AgentTask.last_status == "running", + AgentTask.last_run_id == task.last_run_id, + ).update( + { + "last_status": "interrupted", + "last_result": result, + "updated_at": finished_at, + }, + synchronize_session=False, + )) + + @classmethod + @db_update + def delete_task_and_runs( + cls, + db: Session, + task_id: int, + user_id: Optional[str] = None, + ) -> bool: + """原子删除非运行中任务及其执行历史。""" + query = db.query(AgentTask).filter( + AgentTask.id == task_id, + AgentTask.last_status != "running", + ) + if user_id is not None: + query = query.filter(AgentTask.user_id == user_id) + deleted = query.delete(synchronize_session=False) + if not deleted: + return False + db.query(cls).filter(cls.task_id == task_id).delete(synchronize_session=False) + return True + + @classmethod + @db_query + def get_by_run_id( + cls, + db: Session, + run_id: str, + ) -> Optional["AgentTaskRun"]: + """按公开运行 ID 查询一次执行。""" + return db.query(cls).filter(cls.run_id == run_id).first() + + @classmethod + @db_query + def list_for_task( + cls, + db: Session, + task_id: int, + user_id: Optional[str] = None, + limit: int = 10, + ) -> list["AgentTaskRun"]: + """按父任务 owner 校验后返回最近的有界运行历史。""" + query = db.query(cls).join(AgentTask, AgentTask.id == cls.task_id).filter( + cls.task_id == task_id, + ) + if user_id is not None: + query = query.filter(AgentTask.user_id == user_id) + return query.order_by(cls.started_at.desc(), cls.id.desc()).limit(limit).all() diff --git a/app/scheduler.py b/app/scheduler.py index 26964ba6c..0451e076d 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -1065,7 +1065,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job = self._jobs.get(job_id) if not job or job.get("running"): return False - self.start(job_id) + self.start(job_id, task_id=task_id, trigger_source="manual") return True def init_agent_task_jobs(self) -> None: @@ -1208,17 +1208,25 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): else None ) - async def execute_agent_task(self, task_id: int) -> tuple[bool, str]: + async def execute_agent_task( + self, + task_id: int, + trigger_source: str = "scheduled", + ) -> tuple[bool, str]: """ 唤醒 Agent 执行指定自主定时任务。 :param task_id: Agent 定时任务 ID + :param trigger_source: 触发入口，scheduled-自动调度，manual-显式立即执行 :return: 执行是否成功及结果摘要 """ from app.agent import agent_manager try: - return await agent_manager.execute_scheduled_task(task_id) + return await agent_manager.execute_scheduled_task( + task_id, + trigger_source=trigger_source, + ) finally: task = AgentTaskOper().get(task_id) if task and task.trigger_type == "date" and not task.enabled: diff --git a/database/versions/f4c8d2a7b1e6_3_0_6.py b/database/versions/f4c8d2a7b1e6_3_0_6.py new file mode 100644 index 000000000..29be1526f --- /dev/null +++ b/database/versions/f4c8d2a7b1e6_3_0_6.py @@ -0,0 +1,118 @@ +"""3.0.6 +新增 Agent 自主任务逐次执行记录 + +Revision ID: f4c8d2a7b1e6 +Revises: b3d7e9f1a2c4 +Create Date: 2026-08-13 +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "f4c8d2a7b1e6" +down_revision = "b3d7e9f1a2c4" +branch_labels = None +depends_on = None + + +def _inspector() -> sa.Inspector: + """返回使用当前迁移连接的数据库检查器。""" + return sa.inspect(op.get_bind()) + + +def _has_table(table_name: str) -> bool: + """检查表是否存在。""" + return table_name in _inspector().get_table_names() + + +def _has_column(table_name: str, column_name: str) -> bool: + """检查表字段是否存在。""" + if not _has_table(table_name): + return False + return column_name in { + column["name"] for column in _inspector().get_columns(table_name) + } + + +def _has_index(table_name: str, index_name: str) -> bool: + """检查表索引或唯一约束是否存在。""" + if not _has_table(table_name): + return False + inspector = _inspector() + names = {index.get("name") for index in inspector.get_indexes(table_name)} + names.update( + constraint.get("name") + for constraint in inspector.get_unique_constraints(table_name) + ) + return index_name in names + + +def _id_column(dialect_name: str) -> sa.Column: + """生成与当前 ORM 一致的自增主键定义。""" + if dialect_name == "postgresql": + return sa.Column( + "id", + sa.Integer(), + sa.Identity(start=1, cycle=True), + primary_key=True, + ) + return sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True) + + +def upgrade() -> None: + """建立运行历史表及 AgentTask 最新运行指针。""" + if _has_table("agenttask") and not _has_column("agenttask", "last_run_id"): + with op.batch_alter_table("agenttask") as batch_op: + batch_op.add_column(sa.Column("last_run_id", sa.String(), nullable=True)) + + if not _has_table("agenttaskrun"): + dialect_name = op.get_bind().dialect.name + op.create_table( + "agenttaskrun", + _id_column(dialect_name), + sa.Column("run_id", sa.String(), nullable=False), + sa.Column("task_id", sa.Integer(), nullable=False), + sa.Column("trigger_source", sa.String(), 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()), + sa.Column("run_at", sa.String()), + sa.Column("user_id", sa.String(), nullable=False), + sa.Column("username", sa.String()), + sa.Column("session_id", sa.String(), nullable=False), + sa.Column("channel", sa.String()), + sa.Column("message_source", sa.String()), + sa.Column("original_chat_id", sa.String()), + sa.Column("status", sa.String(), nullable=False), + sa.Column("started_at", sa.String(), nullable=False), + sa.Column("finished_at", sa.String()), + sa.Column("result", sa.Text()), + ) + if not _has_index("agenttaskrun", "ix_agenttaskrun_run_id"): + op.create_index( + "ix_agenttaskrun_run_id", + "agenttaskrun", + ["run_id"], + unique=True, + ) + if not _has_index("agenttaskrun", "ix_agenttaskrun_task_started"): + op.create_index( + "ix_agenttaskrun_task_started", + "agenttaskrun", + ["task_id", "started_at", "id"], + ) + + +def downgrade() -> None: + """删除运行历史并移除 AgentTask 最新运行指针。""" + if _has_table("agenttaskrun"): + if _has_index("agenttaskrun", "ix_agenttaskrun_task_started"): + op.drop_index("ix_agenttaskrun_task_started", table_name="agenttaskrun") + if _has_index("agenttaskrun", "ix_agenttaskrun_run_id"): + op.drop_index("ix_agenttaskrun_run_id", table_name="agenttaskrun") + op.drop_table("agenttaskrun") + if _has_column("agenttask", "last_run_id"): + with op.batch_alter_table("agenttask") as batch_op: + batch_op.drop_column("last_run_id") diff --git a/tests/test_agent_scheduled_tasks.py b/tests/test_agent_scheduled_tasks.py index 6478a2109..24f474f43 100644 --- a/tests/test_agent_scheduled_tasks.py +++ b/tests/test_agent_scheduled_tasks.py @@ -37,7 +37,9 @@ from app.agent.tools.impl.update_agent_task import ( ) from app.agent.tools.tags import ToolTag from app.core.config import settings +from app.db import SessionFactory from app.db.agenttask_oper import AgentTaskOper +from app.db.models.agenttask import AgentTask from app.schemas import ScheduleInfo from app.scheduler import Scheduler from app.utils.timer import TimerUtils @@ -364,7 +366,11 @@ def test_scheduler_restart_keeps_interrupted_date_task_manual_only( scheduler.start = Mock() assert scheduler.start_agent_task(task.id) is True - scheduler.start.assert_called_once_with(job_id) + scheduler.start.assert_called_once_with( + job_id, + task_id=task.id, + trigger_source="manual", + ) @pytest.mark.anyio @@ -381,17 +387,35 @@ async def test_interrupted_date_task_manual_run_disables_and_removes_job( process_message = AsyncMock(return_value="执行完成") monkeypatch.setattr("app.agent.agent_manager.process_message", process_message) - assert await scheduler.execute_agent_task(task.id) == (True, "执行完成") + assert await scheduler.execute_agent_task( + task.id, + trigger_source="manual", + ) == (True, "执行完成") process_message.assert_awaited_once() finished = AgentTaskOper().get(task.id) + runs = AgentTaskOper().list_runs(task.id) job_id = scheduler._get_agent_task_job_id(task.id) assert finished.last_status == "success" assert finished.run_count == 1 assert finished.enabled is False + assert len(runs) == 2 + assert [run.trigger_source for run in runs] == ["manual", "scheduled"] assert job_id not in scheduler._jobs +@pytest.mark.anyio +async def test_scheduler_propagates_scheduled_trigger_source(monkeypatch) -> None: + """自动调度入口应显式保持 scheduled 运行来源。""" + task = _add_agent_task("cron", "0 * * * *", "scheduled-source") + scheduler = _build_agent_task_scheduler() + execute = AsyncMock(return_value=(True, "执行完成")) + monkeypatch.setattr("app.agent.agent_manager.execute_scheduled_task", execute) + + assert await scheduler.execute_agent_task(task.id) == (True, "执行完成") + execute.assert_awaited_once_with(task.id, trigger_source="scheduled") + + @pytest.mark.parametrize("run_time_factory", [_future_time, _invalid_time]) @pytest.mark.anyio async def test_interrupted_date_task_enable_toggle_stays_manual_only( @@ -612,7 +636,11 @@ def test_scheduler_restart_keeps_interrupted_cron_future_schedule() -> None: scheduler.start = Mock() assert scheduler.start_agent_task(task.id) is True - scheduler.start.assert_called_once_with(job_id) + scheduler.start.assert_called_once_with( + job_id, + task_id=task.id, + trigger_source="manual", + ) def test_scheduler_restart_reconciles_disabled_running_agent_task() -> None: @@ -620,7 +648,10 @@ def test_scheduler_restart_reconciles_disabled_running_agent_task() -> None: task = _add_agent_task("cron", "0 * * * *", "restart-disabled") oper = AgentTaskOper() assert oper.mark_running(task.id) - assert oper.update(task_id=task.id, payload={"enabled": False}) + # 冷启动需兼容数据库中的异常状态组合；生产更新入口禁止修改运行中任务。 + with SessionFactory() as db: + db.query(AgentTask).filter(AgentTask.id == task.id).update({"enabled": False}) + db.commit() scheduler = _build_agent_task_scheduler(reconcile=True) scheduler.init_agent_task_jobs() @@ -644,7 +675,11 @@ def test_scheduler_starts_registered_agent_task_without_waiting() -> None: scheduler.start = Mock() assert scheduler.start_agent_task(7) is True - scheduler.start.assert_called_once_with("agent-task-7") + scheduler.start.assert_called_once_with( + "agent-task-7", + task_id=7, + trigger_source="manual", + ) scheduler._jobs["agent-task-7"]["running"] = True assert scheduler.start_agent_task(7) is False @@ -774,8 +809,10 @@ async def test_agent_task_tools_manage_persistent_schedule(monkeypatch) -> None: assert updated["run_at"] is None assert AgentTaskOper().mark_running(task_id) + updated_jobs = list(fake_scheduler.updated) running_update = await update_tool.run(task_id=task_id, enabled=False) assert running_update == f"Agent 定时任务 {task_id} 正在执行，请稍后再修改" + assert fake_scheduler.updated == updated_jobs AgentTaskOper().finish(task_id, success=True, result="完成") run_result = await _build_tool(RunAgentTaskTool, user_id).run(task_id=task_id) diff --git a/tests/test_agent_task_run_migration.py b/tests/test_agent_task_run_migration.py new file mode 100644 index 000000000..7b1046051 --- /dev/null +++ b/tests/test_agent_task_run_migration.py @@ -0,0 +1,106 @@ +import importlib + +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.dialects import postgresql +from sqlalchemy.schema import CreateTable + +from app.db.models.agenttask import AgentTask +from app.db.models.agenttaskrun import AgentTaskRun + + +MIGRATION = "database.versions.f4c8d2a7b1e6_3_0_6" + + +def _bind_migration(monkeypatch, connection): + """把迁移绑定到隔离数据库连接。""" + migration = importlib.import_module(MIGRATION) + context = MigrationContext.configure(connection) + monkeypatch.setattr(migration, "op", Operations(context)) + return migration + + +def _legacy_agent_task(metadata: sa.MetaData) -> None: + """建立迁移前的最小 AgentTask 表。""" + sa.Table( + "agenttask", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("last_status", sa.String(), nullable=False), + ) + + +def test_agent_task_run_migration_upgrades_legacy_schema_and_downgrades( + monkeypatch, +) -> None: + """旧 SQLite schema 应可重复升级并完整回滚新增结构。""" + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + _legacy_agent_task(metadata) + with engine.begin() as connection: + metadata.create_all(connection) + migration = _bind_migration(monkeypatch, connection) + + migration.upgrade() + migration.upgrade() + + inspector = sa.inspect(connection) + assert "agenttaskrun" in inspector.get_table_names() + assert "last_run_id" in { + column["name"] for column in inspector.get_columns("agenttask") + } + indexes = { + index["name"]: (tuple(index["column_names"]), index["unique"]) + for index in inspector.get_indexes("agenttaskrun") + } + assert indexes["ix_agenttaskrun_run_id"] == (("run_id",), 1) + assert indexes["ix_agenttaskrun_task_started"] == ( + ("task_id", "started_at", "id"), + 0, + ) + + migration.downgrade() + inspector = sa.inspect(connection) + assert "agenttaskrun" not in inspector.get_table_names() + assert "last_run_id" not in { + column["name"] for column in inspector.get_columns("agenttask") + } + + +def test_agent_task_run_migration_accepts_fresh_current_schema(monkeypatch) -> None: + """create_all 已建立当前结构时，迁移重复升级不得创建冲突对象。""" + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + AgentTask.__table__.create(connection) + AgentTaskRun.__table__.create(connection) + migration = _bind_migration(monkeypatch, connection) + + migration.upgrade() + migration.upgrade() + + inspector = sa.inspect(connection) + assert "agenttaskrun" in inspector.get_table_names() + assert { + column["name"] for column in inspector.get_columns("agenttaskrun") + } == {column.name for column in AgentTaskRun.__table__.columns} + assert len(inspector.get_indexes("agenttaskrun")) == 2 + + +def test_agent_task_run_migration_matches_postgresql_identity() -> None: + """独立 Alembic 路径应与 PostgreSQL create_all 使用相同 Identity。""" + migration = importlib.import_module(MIGRATION) + metadata = sa.MetaData() + table = sa.Table( + "agenttaskrun", + metadata, + migration._id_column("postgresql"), + ) + + identity = table.c.id.identity + assert identity is not None + assert identity.start == 1 + assert identity.cycle is True + ddl = str(CreateTable(table).compile(dialect=postgresql.dialect())) + assert "GENERATED BY DEFAULT AS IDENTITY" in ddl + assert "CYCLE" in ddl diff --git a/tests/test_agent_task_runs.py b/tests/test_agent_task_runs.py new file mode 100644 index 000000000..2a750f3c5 --- /dev/null +++ b/tests/test_agent_task_runs.py @@ -0,0 +1,353 @@ +import json +from concurrent.futures import ThreadPoolExecutor +from threading import Event, Thread, current_thread +from uuid import uuid4 + +import pytest +from sqlalchemy import event +from sqlalchemy.exc import IntegrityError + +from app.agent import AgentManager +from app.agent.tools.impl.query_agent_tasks import QueryAgentTasksTool +from app.db import Engine, SessionFactory +from app.db.agenttask_oper import AgentTaskOper +from app.db.models.agenttask import AgentTask +from app.db.models.agenttaskrun import AgentTaskRun + + +def _add_task(prefix: str, *, trigger_type: str = "cron") -> AgentTask: + """创建带隔离 owner 的 Agent 自主任务。""" + user_id = f"{prefix}-{uuid4().hex}" + return AgentTaskOper().add( + name=f"{prefix} 检查", + content="检查资源并报告", + trigger_type=trigger_type, + cron_expression="0 * * * *" if trigger_type == "cron" else None, + run_at="2099-01-01T00:00:00+08:00" if trigger_type == "date" else None, + user_id=user_id, + username="admin", + session_id=f"session-{user_id}", + channel="Telegram", + source="telegram-test", + original_chat_id="chat-1", + ) + + +def _build_query_tool(user_id: str) -> QueryAgentTasksTool: + """构造绑定当前 owner 的任务查询工具。""" + tool = QueryAgentTasksTool(session_id=f"session-{user_id}", user_id=user_id) + tool._message_context = {"username": "admin"} + return tool + + +def test_begin_run_claims_once_and_preserves_snapshot() -> None: + """并发认领只能创建一个 run，且任务修改不改变执行快照。""" + task = _add_task("run-claim") + + with ThreadPoolExecutor(max_workers=2) as executor: + runs = list(executor.map( + lambda source: AgentTaskOper().begin_run(task.id, source), + ("scheduled", "manual"), + )) + + created = [run for run in runs if run] + assert len(created) == 1 + run = created[0] + assert run.trigger_source in {"scheduled", "manual"} + assert run.name == task.name + assert run.content == task.content + assert not AgentTaskOper().update(task.id, {"name": "运行中不可修改"}) + assert AgentTaskOper().finish_run(run.run_id, success=True, result="完成") + assert AgentTaskOper().update(task.id, {"name": "新名称", "content": "新内容"}) + + snapshot = AgentTaskOper().get_run(run.run_id) + current = AgentTaskOper().get(task.id) + assert snapshot.name == task.name + assert snapshot.content == task.content + assert current.name == "新名称" + assert current.content == "新内容" + assert current.last_run_id == run.run_id + + +def test_begin_run_uses_configuration_committed_before_atomic_claim(monkeypatch) -> None: + """配置先完成写入时，执行快照不得因秒级时间相同而读取旧值。""" + fixed_time = "2026-08-13 20:00:00" + monkeypatch.setattr(AgentTaskOper, "_now", staticmethod(lambda: fixed_time)) + task = _add_task("run-current-snapshot") + claim_ready = Event() + update_done = Event() + result = {} + + def pause_before_claim( + _connection, + _cursor, + statement, + _parameters, + _context, + _executemany, + ) -> None: + if ( + current_thread().name == "agent-task-claim" + and statement.lstrip().upper().startswith("UPDATE AGENTTASK SET") + ): + claim_ready.set() + assert update_done.wait(timeout=5) + + def begin() -> None: + result["run"] = AgentTaskOper().begin_run(task.id) + + event.listen(Engine, "before_cursor_execute", pause_before_claim) + try: + thread = Thread(target=begin, name="agent-task-claim") + thread.start() + assert claim_ready.wait(timeout=5) + assert AgentTaskOper().update( + task.id, + {"name": "最新名称", "content": "最新内容"}, + ) + update_done.set() + thread.join(timeout=5) + assert not thread.is_alive() + finally: + update_done.set() + event.remove(Engine, "before_cursor_execute", pause_before_claim) + + run = result["run"] + assert run.name == "最新名称" + assert run.content == "最新内容" + + +def test_begin_run_rejects_unknown_trigger_source() -> None: + """运行记录只接受已定义的定时或手动触发入口。""" + task = _add_task("run-source") + + with pytest.raises(ValueError, match="不支持的 Agent 任务触发来源"): + AgentTaskOper().begin_run(task.id, "retry") + + unchanged = AgentTaskOper().get(task.id) + assert unchanged.last_status == "waiting" + assert unchanged.last_run_id is None + assert AgentTaskOper().list_runs(task.id) == [] + + +def test_begin_run_rolls_back_task_claim_when_run_insert_fails() -> None: + """运行记录插入失败时，任务的 running 投影必须随事务回滚。""" + first_task = _add_task("run-rollback-first") + second_task = _add_task("run-rollback-second") + run_id = uuid4().hex + assert AgentTaskRun.begin_run( + None, + task_id=first_task.id, + run_id=run_id, + trigger_source="scheduled", + started_at="2026-08-13 20:00:00", + ) == run_id + + with pytest.raises(IntegrityError): + AgentTaskRun.begin_run( + None, + task_id=second_task.id, + run_id=run_id, + trigger_source="manual", + started_at="2026-08-13 20:00:01", + ) + + unchanged = AgentTaskOper().get(second_task.id) + assert unchanged.last_status == "waiting" + assert unchanged.last_run_id is None + assert len(AgentTaskOper().list_runs(first_task.id)) == 1 + assert AgentTaskOper().list_runs(second_task.id) == [] + + +def test_finish_run_finalizes_once_under_concurrency() -> None: + """同一 run 的并发收口只能有一个成功并只累计一次。""" + task = _add_task("run-finish-once") + run = AgentTaskOper().begin_run(task.id) + assert run + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map( + lambda value: AgentTaskOper().finish_run( + run.run_id, + success=True, + result=value, + ), + ("结果 A", "结果 B"), + )) + + assert sorted(results) == [False, True] + completed = AgentTaskOper().get(task.id) + finalized = AgentTaskOper().get_run(run.run_id) + assert completed.last_status == "success" + assert completed.last_result in {"结果 A", "结果 B"} + assert completed.run_count == 1 + assert finalized.status == "success" + assert finalized.result == completed.last_result + + +def test_stale_finish_cannot_overwrite_latest_run_projection() -> None: + """迟到的旧运行只能收口自己，不得覆盖任务的最新运行投影。""" + task = _add_task("run-stale") + oper = AgentTaskOper() + first = oper.begin_run(task.id, "scheduled") + assert first + + with SessionFactory() as db: + db.query(AgentTask).filter(AgentTask.id == task.id).update({ + "last_status": "interrupted", + }) + db.commit() + second = oper.begin_run(task.id, "manual") + assert second + + assert oper.finish_run(first.run_id, success=True, result="旧结果") + current = oper.get(task.id) + assert current.last_run_id == second.run_id + assert current.last_status == "running" + assert current.last_result is None + assert current.run_count == 0 + assert oper.get_run(first.run_id).status == "success" + + assert oper.finish_run(second.run_id, success=False, result="新结果") + finished = oper.get(task.id) + assert finished.last_status == "failed" + assert finished.last_result == "新结果" + assert finished.run_count == 1 + + +def test_interruption_requires_matching_running_run() -> None: + """有 run 指针时，对账不得只改任务而留下不一致的运行历史。""" + task = _add_task("run-interrupt-mismatch") + oper = AgentTaskOper() + run = oper.begin_run(task.id) + assert run + with SessionFactory() as db: + db.query(AgentTaskRun).filter(AgentTaskRun.run_id == run.run_id).update({ + "status": "success", + "result": "已收口", + }) + db.commit() + + assert not oper.mark_interrupted(task.id, "不得覆盖") + unchanged = oper.get(task.id) + assert unchanged.last_status == "running" + assert unchanged.last_result is None + assert oper.get_run(run.run_id).status == "success" + + +def test_interruption_supports_legacy_running_task_without_run() -> None: + """升级前遗留的 running 投影没有 run 指针时仍需兼容对账。""" + task = _add_task("run-interrupt-legacy") + with SessionFactory() as db: + db.query(AgentTask).filter(AgentTask.id == task.id).update({ + "last_status": "running", + "last_run_id": None, + }) + db.commit() + + oper = AgentTaskOper() + assert oper.mark_interrupted(task.id, "旧任务结果未知") + interrupted = oper.get(task.id) + assert interrupted.last_status == "interrupted" + assert interrupted.last_result == "旧任务结果未知" + assert interrupted.last_run_id is None + assert oper.list_runs(task.id) == [] + + +def test_interruption_and_manual_rerun_keep_distinct_history() -> None: + """中断对账与显式重跑应保留两条互不覆盖的执行记录。""" + task = _add_task("run-interrupt", trigger_type="date") + oper = AgentTaskOper() + first = oper.begin_run(task.id, "scheduled") + assert first + assert oper.mark_interrupted(task.id, "执行结果未知") + assert oper.get_run(first.run_id).status == "interrupted" + + second = oper.begin_run(task.id, "manual") + assert second and second.run_id != first.run_id + assert oper.finish_run( + second.run_id, + success=True, + result="重跑完成", + disable_date_task=True, + ) + + runs = oper.list_runs(task.id) + assert [run.run_id for run in runs] == [second.run_id, first.run_id] + assert [run.status for run in runs] == ["success", "interrupted"] + finished = oper.get(task.id) + assert finished.enabled is False + assert finished.last_status == "success" + assert finished.run_count == 1 + + +def test_delete_rejects_running_task_and_removes_all_history() -> None: + """运行中任务不可删除，收口后永久删除不得留下孤立 run。""" + task = _add_task("run-delete") + oper = AgentTaskOper() + run = oper.begin_run(task.id) + assert run + assert not oper.delete(task.id, user_id=task.user_id) + assert oper.get(task.id) is not None + assert oper.finish_run(run.run_id, success=True, result="完成") + + assert not oper.delete(task.id, user_id="other-user") + assert oper.delete(task.id, user_id=task.user_id) + assert oper.get(task.id) is None + assert oper.get_run(run.run_id) is None + + +@pytest.mark.anyio +async def test_query_task_returns_owner_scoped_ten_recent_runs(monkeypatch) -> None: + """单任务查询只向 owner 返回最近十次运行，列表查询不携带历史。""" + task = _add_task("run-query") + other = _add_task("run-query-other") + oper = AgentTaskOper() + expected = [] + for index in range(12): + run = oper.begin_run(task.id, "manual" if index % 2 else "scheduled") + assert run + assert oper.finish_run(run.run_id, success=True, result=f"结果 {index}") + expected.insert(0, run.run_id) + other_run = oper.begin_run(other.id) + assert other_run + assert oper.finish_run(other_run.run_id, success=True, result="其他用户") + + monkeypatch.setattr( + "app.scheduler.Scheduler.get_agent_task_next_run", + lambda _self, _task_id: None, + ) + detail = json.loads(await _build_query_tool(task.user_id).run(task_id=task.id)) + assert detail["total"] == 1 + assert [run["run_id"] for run in detail["tasks"][0]["recent_runs"]] == expected[:10] + assert all(run["task_id"] == task.id for run in detail["tasks"][0]["recent_runs"]) + + listing = json.loads(await _build_query_tool(task.user_id).run()) + assert listing["total"] == 1 + assert "recent_runs" not in listing["tasks"][0] + hidden = json.loads(await _build_query_tool(other.user_id).run(task_id=task.id)) + assert hidden == {"total": 0, "tasks": []} + + +@pytest.mark.anyio +async def test_agent_manager_records_manual_trigger_source(monkeypatch) -> None: + """真实执行入口应把手动触发来源写入对应 run。""" + monkeypatch.setattr("app.agent.settings.AI_AGENT_ENABLE", True) + task = _add_task("run-manager") + manager = AgentManager() + captured = {} + + async def process_message(**kwargs): + captured.update(kwargs) + return "完成" + + manager.process_message = process_message + assert await manager.execute_scheduled_task(task.id, trigger_source="manual") == ( + True, + "完成", + ) + runs = AgentTaskOper().list_runs(task.id) + assert len(runs) == 1 + assert runs[0].trigger_source == "manual" + assert runs[0].status == "success" + assert "定时任务已手动触发" in captured["message"]