diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 97e41e24e..9fd8481c3 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -79,7 +79,7 @@ from app.application.plugin.runtime import get_plugin_manager def _get_plugin_tools_revision() -> int: """读取插件工具目录修订号，避免 Agent 编排依赖具体管理器类型。""" return get_plugin_manager().get_plugin_agent_tools_revision() -from app.application.agentdata import get_agent_task_port +from app.application.agenttask import get_agent_task_execution_service from app.application.agentdata import get_agent_user_port from app.application.messaging.chat import ( get_configured_agent_chat_service, @@ -3504,6 +3504,8 @@ class AgentManager: self, task_id: int, trigger_source: str = "scheduled", + scheduler_generation: int | None = None, + remove_schedule: Callable[[int, int, str], bool] | None = None, ) -> tuple[bool, str]: """ 按持久化上下文唤醒 Agent 执行自主定时任务并向用户回传结果。 @@ -3514,13 +3516,17 @@ class AgentManager: """ if not settings.AI_AGENT_ENABLE: return False, "AI Agent 未启用" - oper = get_agent_task_port() - task = oper.get(task_id) - if not task or not task.enabled: - return False, "Agent 定时任务不存在或已停用" - run = oper.begin_run(task_id=task_id, trigger_source=trigger_source) - if not run: - return False, "Agent 定时任务当前不可执行" + accepting_before_claim = self._accepting_tasks + task_service = get_agent_task_execution_service() + claim = await task_service.claim( + task_id=task_id, + trigger_source=trigger_source, + scheduler_generation=scheduler_generation, + remove_schedule=remove_schedule, + ) + run = claim.run + if run is None: + return False, claim.rejection or "Agent 定时任务当前不可执行" trigger_description = ( "已手动触发" if run.trigger_source == "manual" else "已按计划触发" @@ -3558,7 +3564,14 @@ class AgentManager: raise except Exception as err: success = False - result = f"Agent 定时任务执行失败：{str(err)}" + error_message = str(err) + if ( + accepting_before_claim + and not self._accepting_tasks + and error_message == "AgentManager 未运行或已关闭" + ): + error_message = "AgentManager 已关闭" + result = f"Agent 定时任务执行失败：{error_message}" logger.error(f"Agent 定时任务 {task_id} 执行失败: {str(err)}") await AgentChain().async_post_message( Message( @@ -3570,11 +3583,12 @@ class AgentManager: ) ) finally: - oper.finish_run( - run_id=run.run_id, + await task_service.finalize( + run, success=success, result=str(result or ""), - disable_date_task=run.trigger_type == "date", + scheduler_generation=scheduler_generation, + remove_schedule=remove_schedule, ) return success, str(result or "任务执行完成") diff --git a/app/application/agenttask.py b/app/application/agenttask.py new file mode 100644 index 000000000..497a6827d --- /dev/null +++ b/app/application/agenttask.py @@ -0,0 +1,290 @@ +"""Agent 自主定时任务执行的异步应用边界。""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from dataclasses import dataclass +from typing import Protocol, TypeVar +from uuid import uuid4 + +from app.application.database import AsyncDatabaseExecutor +from app.runtime.execution import await_task_to_terminal +from app.schemas.exception import DatabaseWorkerOverloadedError + + +T = TypeVar("T") +SyncTransaction = Callable[[Callable[[object], T]], T] + + +class AgentTaskRecord(Protocol): + """执行认领与终态判定所需的任务投影。""" + + id: int + enabled: bool + last_run_id: str | None + last_status: str + + +class AgentTaskRunRecord(Protocol): + """执行期间需要脱离数据库会话持有的运行记录字段。""" + + run_id: str + task_id: int + trigger_source: str + name: str + content: str + trigger_type: str + cron_expression: str | None + run_at: str | None + user_id: str + username: str | None + session_id: str + + +class AgentTaskRepository(Protocol): + """AgentTask 执行用例使用的同步短事务仓储合同。""" + + def get(self, task_id: int) -> AgentTaskRecord | None: + """读取任务当前投影。""" + + def begin_run( + self, + task_id: int, + trigger_source: str = "scheduled", + *, + run_id: str | None = None, + ) -> AgentTaskRunRecord | None: + """原子认领任务并创建运行快照。""" + + def finish_run_outcome( + self, + run_id: str, + success: bool, + result: str, + ) -> AgentTaskFinishRecord: + """收口运行并返回事务确认的终态事实。""" + + +class AgentTaskFinishRecord(Protocol): + """同步仓储返回的结构化运行终态。""" + + run_finalized: bool + task_projection_updated: bool + date_task_disabled: bool + + +AgentTaskRepositoryFactory = Callable[[object], AgentTaskRepository] +AgentTaskScheduleRemover = Callable[[int, int, str], bool] + + +@dataclass(frozen=True, slots=True) +class AgentTaskRunSnapshot: + """任务认领成功后可安全跨越数据库会话的执行快照。""" + + run_id: str + task_id: int + trigger_source: str + name: str + content: str + trigger_type: str + cron_expression: str | None + run_at: str | None + user_id: str + username: str | None + session_id: str + + +@dataclass(frozen=True, slots=True) +class AgentTaskClaim: + """任务认领结果；拒绝原因保持现有 Agent 用户提示合同。""" + + run: AgentTaskRunSnapshot | None + rejection: str | None = None + + +@dataclass(frozen=True, slots=True) +class AgentTaskFinishOutcome: + """区分运行收口、任务投影更新与一次任务停用三个事实。""" + + run_finalized: bool + task_projection_updated: bool + date_task_disabled: bool + + +class AgentTaskExecutionService: + """通过有界数据库 worker 认领并收口一次 AgentTask 执行。""" + + def __init__( + self, + *, + repository: AgentTaskRepositoryFactory, + async_executor: AsyncDatabaseExecutor, + sync_transaction: SyncTransaction, + ) -> None: + """保存同步仓储、事务和异步执行器。""" + self._repository = repository + self._async_executor = async_executor + self._sync_transaction = sync_transaction + + @staticmethod + def _snapshot(run: AgentTaskRunRecord) -> AgentTaskRunSnapshot: + """在事务内复制运行字段，避免 ORM 对象越过会话边界。""" + return AgentTaskRunSnapshot( + 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, + user_id=run.user_id, + username=run.username, + session_id=run.session_id, + ) + + async def claim( + self, + task_id: int, + trigger_source: str = "scheduled", + *, + scheduler_generation: int | None = None, + remove_schedule: AgentTaskScheduleRemover | None = None, + ) -> AgentTaskClaim: + """认领一次执行；取消发生在提交后时先补偿收口再传播取消。""" + + run_id = uuid4().hex + + def transaction(session: object) -> AgentTaskClaim: + repository = self._repository(session) + run = repository.begin_run( + task_id=task_id, + trigger_source=trigger_source, + run_id=run_id, + ) + if not run: + task = repository.get(task_id) + return AgentTaskClaim( + run=None, + rejection=( + "Agent 定时任务不存在或已停用" + if not task or not task.enabled + else "Agent 定时任务当前不可执行" + ), + ) + return AgentTaskClaim(run=self._snapshot(run)) + + async def claim_to_terminal() -> AgentTaskClaim: + """容量瞬时耗尽时保留本轮调度，直到认领取得 admission。""" + while True: + try: + return await self._async_executor.run( + lambda: self._sync_transaction(transaction) + ) + except DatabaseWorkerOverloadedError: + await asyncio.sleep(0.01) + + claim_task = asyncio.create_task(claim_to_terminal()) + try: + return await claim_task + except asyncio.CancelledError as cancellation: + finalize_task = asyncio.create_task(self._finalize( + run_id=run_id, + task_id=task_id, + success=False, + result="Agent 定时任务已取消", + scheduler_generation=scheduler_generation, + remove_schedule=remove_schedule, + )) + await await_task_to_terminal(finalize_task) + raise cancellation + + async def finalize( + self, + run: AgentTaskRunSnapshot, + *, + success: bool, + result: str, + scheduler_generation: int | None = None, + remove_schedule: AgentTaskScheduleRemover | None = None, + ) -> AgentTaskFinishOutcome: + """等待终态事务完成，并仅清理仍属于该 generation 的一次任务。""" + + return await self._finalize( + run_id=run.run_id, + task_id=run.task_id, + success=success, + result=result, + scheduler_generation=scheduler_generation, + remove_schedule=remove_schedule, + ) + + async def _finalize( + self, + *, + run_id: str, + task_id: int, + success: bool, + result: str, + scheduler_generation: int | None, + remove_schedule: AgentTaskScheduleRemover | None, + ) -> AgentTaskFinishOutcome: + """按稳定运行 ID 收口，供正常路径和取消补偿共享。""" + + def transaction(session: object) -> AgentTaskFinishOutcome: + repository = self._repository(session) + outcome = repository.finish_run_outcome( + run_id=run_id, + success=success, + result=result, + ) + return AgentTaskFinishOutcome( + run_finalized=outcome.run_finalized, + task_projection_updated=outcome.task_projection_updated, + date_task_disabled=outcome.date_task_disabled, + ) + + async def finish_to_terminal() -> AgentTaskFinishOutcome: + """容量瞬时耗尽时保留 owner，直到收口取得 admission。""" + while True: + try: + return await self._async_executor.run( + lambda: self._sync_transaction(transaction) + ) + except DatabaseWorkerOverloadedError: + await asyncio.sleep(0.01) + + finish_task = asyncio.create_task(finish_to_terminal()) + cancellation: asyncio.CancelledError | None = None + try: + outcome = await asyncio.shield(finish_task) + except asyncio.CancelledError as error: + cancellation = error + outcome = await await_task_to_terminal(finish_task) + + if ( + outcome.date_task_disabled + and scheduler_generation is not None + and remove_schedule is not None + ): + remove_schedule(task_id, scheduler_generation, run_id) + if cancellation is not None: + raise cancellation + return outcome + + +_service: AgentTaskExecutionService | None = None + + +def configure_agent_task_execution(service: AgentTaskExecutionService) -> None: + """由启动组合根登记 AgentTask 执行服务。""" + global _service + _service = service + + +def get_agent_task_execution_service() -> AgentTaskExecutionService: + """返回已登记的 AgentTask 执行服务。""" + if _service is None: + raise RuntimeError("AgentTask 执行服务尚未配置") + return _service diff --git a/app/db/oper/agenttask.py b/app/db/oper/agenttask.py index e21eb858b..475e21dab 100644 --- a/app/db/oper/agenttask.py +++ b/app/db/oper/agenttask.py @@ -1,9 +1,11 @@ from __future__ import annotations +from dataclasses import dataclass from datetime import datetime from typing import Optional from uuid import uuid4 +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session @@ -16,6 +18,15 @@ from app.db.models.agenttask import ( from app.db.models.agenttaskrun import AgentTaskRun +@dataclass(frozen=True, slots=True) +class AgentTaskFinishRecord: + """一次运行收口后由数据库事务确认的三个独立事实。""" + + run_finalized: bool + task_projection_updated: bool + date_task_disabled: bool + + class AgentTaskOper(DbOper): """ Agent 自主定时任务管理。 @@ -238,6 +249,62 @@ class AgentTaskOper(DbOper): ) ) + def finish_run_outcome( + self, + run_id: str, + success: bool, + result: str, + ) -> AgentTaskFinishRecord: + """收口运行，并返回当前事务实际更新的任务投影和停用事实。""" + finished_at = self._now() + normalized_result = (result or "")[:20000] + expected_status = "success" if success else "failed" + + def finalize(session: Session) -> AgentTaskFinishRecord: + """使用列查询绕过 ORM identity-map，读取刚写入的真实投影。""" + finalized = AgentTaskRun.finish_run( + session, + run_id=run_id, + success=success, + result=normalized_result, + finished_at=finished_at, + disable_date_task=True, + ) + run = session.execute( + select( + AgentTaskRun.task_id, + AgentTaskRun.trigger_type, + ).where(AgentTaskRun.run_id == run_id) + ).mappings().first() + task = None + if run: + task = session.execute( + select( + AgentTask.last_run_id, + AgentTask.last_status, + AgentTask.enabled, + ).where(AgentTask.id == run["task_id"]) + ).mappings().first() + projection_updated = bool( + finalized + and task + and task["last_run_id"] == run_id + and task["last_status"] == expected_status + ) + return AgentTaskFinishRecord( + run_finalized=finalized, + task_projection_updated=projection_updated, + date_task_disabled=bool( + projection_updated + and run + and run["trigger_type"] == "date" + and task + and not task["enabled"] + ), + ) + + return self._execute_sync_write(finalize) + def finish( self, task_id: int, diff --git a/app/scheduler.py b/app/scheduler.py index d9be86097..72f47e7f2 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -28,7 +28,7 @@ from app.chain.transfer import TransferChain from app.chain.workflow import WorkflowChain from app.runtime.config import global_vars from app.runtime.events import Event, eventmanager -from app.db.oper.agenttask import AgentTaskOper +from app.application.agentdata import get_agent_task_port from app.application.database import get_database_governance from app.application.outbox import dispatch_pending_outbox from app.application.plugin.runtime import get_plugin_manager @@ -1274,6 +1274,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): func = job.get("func") if not func: return + if func == self.execute_agent_task: + kwargs.setdefault("scheduler_generation", generation) if self.__supports_progress_callback(func) and "progress_callback" not in kwargs: kwargs["progress_callback"] = self.__build_progress_callback( job_id=job_id, job=job @@ -1426,7 +1428,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """ 按数据库当前状态注册所有启用的 Agent 自主定时任务。 """ - for task in AgentTaskOper().list(enabled=True): + for task in get_agent_task_port().list(enabled=True): self.update_agent_task_job(task.id) def _reconcile_agent_task_interruptions(self) -> None: @@ -1439,7 +1441,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): with self._lock: if self._agent_task_interruptions_reconciled: return - oper = AgentTaskOper() + oper = get_agent_task_port() for task in oper.list(): if task.last_status == "running": oper.mark_interrupted( @@ -1460,7 +1462,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """ config = get_scheduler_runtime_config() self.remove_agent_task_job(task_id) - task = AgentTaskOper().get(task_id) + task = get_agent_task_port().get(task_id) if ( not config.ai_agent_enable or not task @@ -1496,6 +1498,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): kwargs={"task_id": task_id}, ).to_runtime_state() self._assign_job_generation(job_id, job) + job["_agent_task_run_id"] = task.last_run_id + job["_agent_task_status"] = task.last_status self._jobs[job_id] = job self._jobs[job_id]["provider_name"] = "[Agent]" # 已开始的一次任务在重启后结果未知，只保留显式执行入口，不能按 @@ -1531,6 +1535,31 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): except JobLookupError: pass + def _remove_agent_task_job_generation( + self, + task_id: int, + generation: int, + run_id: str, + ) -> bool: + """移除本次执行或其运行中重载产生的 AgentTask 调度注册。""" + job_id = self._get_agent_task_job_id(task_id) + with self._lock: + job = self._jobs.get(job_id) + if job is None: + return False + if job.get("_generation", 0) != generation and not ( + job.get("_agent_task_run_id") == run_id + and job.get("_agent_task_status") == "running" + ): + return False + self._jobs.pop(job_id, None) + if self._scheduler: + try: + self._scheduler.remove_job(job_id) + except JobLookupError: + pass + return True + def get_agent_task_next_run(self, task_id: int) -> Optional[str]: """ 查询 Agent 自主定时任务的下一次执行时间。 @@ -1546,7 +1575,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): if next_run_time: return next_run_time.isoformat(timespec="seconds") - task = AgentTaskOper().get(task_id) + task = get_agent_task_port().get(task_id) if not task or not task.enabled: return None if task.trigger_type == "date" and task.last_status == "interrupted": @@ -1572,6 +1601,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): self, task_id: int, trigger_source: str = "scheduled", + scheduler_generation: int | None = None, ) -> tuple[bool, str]: """ 唤醒 Agent 执行指定自主定时任务。 @@ -1582,19 +1612,23 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """ from app.application.agent import get_running_agent_manager - try: - manager = get_running_agent_manager() - if manager is None: - logger.warning("智能助手服务未运行，跳过 Agent 定时任务") - return False, "智能助手服务未运行" - return await manager.execute_scheduled_task( - task_id, - trigger_source=trigger_source, + manager = get_running_agent_manager() + if manager is None: + logger.warning("智能助手服务未运行，跳过 Agent 定时任务") + return False, "智能助手服务未运行" + if scheduler_generation is None: + job_id = self._get_agent_task_job_id(task_id) + with self._lock: + job = self._jobs.get(job_id) + if job is not None: + scheduler_generation = job.get("_generation", 0) + kwargs: dict[str, Any] = {"trigger_source": trigger_source} + if scheduler_generation is not None: + kwargs.update( + scheduler_generation=scheduler_generation, + remove_schedule=self._remove_agent_task_job_generation, ) - finally: - task = await AgentTaskOper().async_get(task_id) - if task and task.trigger_type == "date" and not task.enabled: - self.remove_agent_task_job(task_id) + return await manager.execute_scheduled_task(task_id, **kwargs) def init_plugin_jobs(self): """ diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index bad01bf15..2977a37fe 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -95,6 +95,10 @@ from app.application.site.query import SiteQueryService, configure_site_query_se from app.application.site.health import SiteHealthService, configure_site_health_service from app.application.workflow import WorkflowQueryService, configure_workflow_query from app.application.agentdata import configure_agent_data_ports +from app.application.agenttask import ( + AgentTaskExecutionService, + configure_agent_task_execution, +) from app.api.data import ApiDataPorts, configure_api_data_runtime from app.application.subscription.write import configure_subscribe_writer from app.adapters.external.server import ( @@ -831,6 +835,11 @@ async def init_modules() -> HostRuntime: workflow=lambda: WorkflowOper(), plugin_data=lambda: PluginDataOper(), ) + configure_agent_task_execution(AgentTaskExecutionService( + repository=lambda session: AgentTaskOper(session), + async_executor=database_worker, + sync_transaction=transaction_runner.sync, + )) configure_subscribe_writer( lambda: TransactionalSubscribeWriter( sync_session=SessionFactory, diff --git a/tests/conftest.py b/tests/conftest.py index e72136d6a..f29583aa4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -153,6 +153,10 @@ def configure_plugin_system_services(): from app.workflow import WorkFlowManager configure_workflow_runtime(lambda: WorkFlowManager()) from app.application.agentdata import configure_agent_data_ports + from app.application.agenttask import ( + AgentTaskExecutionService, + configure_agent_task_execution, + ) from app.db.oper.agentchat import AgentChatOper from app.db.oper.downloadfailure import DownloadFailureOper from app.db.oper.downloadhistory import DownloadHistoryOper @@ -272,6 +276,11 @@ def configure_plugin_system_services(): workflow=lambda: WorkflowOper(), plugin_data=lambda: PluginDataOper(), ) + configure_agent_task_execution(AgentTaskExecutionService( + repository=lambda session: AgentTaskOper(session), + async_executor=database_executor, + sync_transaction=transaction_runner.sync, + )) configure_agent_chat_persistence( AgentChatPersistenceService( repository=lambda session: AgentChatOper(session), diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 417fa9aa6..701b27e92 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6554, - "edge_sha256": "1c72744be67d95f98eac1c7b1a72bdbf7b6d1d4f337556b8e22293c3a9525c67", + "edge_count": 6560, + "edge_sha256": "bdd34affb7c42a4cbcdc85e713d9b4ac591c3343559dd5032b4518e77eba721e", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -356,6 +356,7 @@ "app.agent.orchestrator -> app.agent.tools.impl.query_system_settings", "app.agent.orchestrator -> app.application", "app.agent.orchestrator -> app.application.agentdata", + "app.agent.orchestrator -> app.application.agenttask", "app.agent.orchestrator -> app.application.messaging", "app.agent.orchestrator -> app.application.messaging.chat", "app.agent.orchestrator -> app.application.plugin", @@ -2483,6 +2484,12 @@ "app.api.servcookie -> app.runtime.log", "app.api.servcookie -> app.schemas", "app.api.servcookie -> app.schemas.servcookie", + "app.application.agenttask -> app.application", + "app.application.agenttask -> app.application.database", + "app.application.agenttask -> app.runtime", + "app.application.agenttask -> app.runtime.execution", + "app.application.agenttask -> app.schemas", + "app.application.agenttask -> app.schemas.exception", "app.application.audio -> app.domain", "app.application.audio -> app.domain.context", "app.application.audio -> app.domain.meta", @@ -5809,6 +5816,7 @@ "app.scheduler -> app.adapters.external.server", "app.scheduler -> app.application", "app.scheduler -> app.application.agent", + "app.scheduler -> app.application.agentdata", "app.scheduler -> app.application.configuration", "app.scheduler -> app.application.database", "app.scheduler -> app.application.image", @@ -5827,9 +5835,6 @@ "app.scheduler -> app.chain.subscribe", "app.scheduler -> app.chain.transfer", "app.scheduler -> app.chain.workflow", - "app.scheduler -> app.db", - "app.scheduler -> app.db.oper", - "app.scheduler -> app.db.oper.agenttask", "app.scheduler -> app.foundation", "app.scheduler -> app.foundation.singleton", "app.scheduler -> app.runtime", @@ -6187,6 +6192,7 @@ "app.startup.initializers.modules -> app.api.data", "app.startup.initializers.modules -> app.application", "app.startup.initializers.modules -> app.application.agentdata", + "app.startup.initializers.modules -> app.application.agenttask", "app.startup.initializers.modules -> app.application.chain", "app.startup.initializers.modules -> app.application.chain.context", "app.startup.initializers.modules -> app.application.chain.data", @@ -6571,7 +6577,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 809, + "module_count": 810, "modules": [ "app", "app.adapters", @@ -6820,6 +6826,7 @@ "app.application", "app.application.agent", "app.application.agentdata", + "app.application.agenttask", "app.application.audio", "app.application.backup", "app.application.chain", diff --git a/tests/test_agent_scheduled_tasks.py b/tests/test_agent_scheduled_tasks.py index 3ad3f77fc..8c32bddc8 100644 --- a/tests/test_agent_scheduled_tasks.py +++ b/tests/test_agent_scheduled_tasks.py @@ -352,6 +352,108 @@ def test_scheduler_registers_and_removes_agent_task_job() -> None: scheduler._scheduler.shutdown(wait=False) +def test_stale_agent_task_generation_cannot_remove_replacement_job() -> None: + """旧执行收尾不得删除配置刷新后注册的新 generation。""" + task = _add_agent_task("date", _future_time(), "generation-replace") + scheduler = _build_agent_task_scheduler() + scheduler.update_agent_task_job(task.id) + job_id = scheduler._get_agent_task_job_id(task.id) + old_generation = scheduler._jobs[job_id]["_generation"] + + scheduler.update_agent_task_job(task.id) + new_generation = scheduler._jobs[job_id]["_generation"] + + assert new_generation > old_generation + assert scheduler._remove_agent_task_job_generation( + task.id, + old_generation, + "old-run", + ) is False + assert scheduler._jobs[job_id]["_generation"] == new_generation + assert scheduler._scheduler.get_job(job_id) is not None + + +@pytest.mark.anyio +async def test_date_task_reload_job_is_removed_after_run_finishes( + monkeypatch, +) -> None: + """运行中重载生成的同一次任务副本必须随 date 终态一起移除。""" + task = _add_agent_task("date", _future_time(), "date-reload-active") + scheduler = _build_agent_task_scheduler() + scheduler.update_agent_task_job(task.id) + job_id = scheduler._get_agent_task_job_id(task.id) + original_generation = scheduler._jobs[job_id]["_generation"] + started = asyncio.Event() + release = asyncio.Event() + + async def process_message(**_kwargs) -> str: + started.set() + await release.wait() + return "执行完成" + + manager = SimpleNamespace( + execute_scheduled_task=AgentManager.execute_scheduled_task, + process_message=process_message, + _accepting_tasks=True, + ) + manager.execute_scheduled_task = AgentManager.execute_scheduled_task.__get__(manager) + monkeypatch.setattr( + "app.application.agent.get_running_agent_manager", + lambda: manager, + ) + + assert scheduler.start(job_id) is True + await asyncio.wait_for(started.wait(), timeout=1) + scheduler.update_agent_task_job(task.id) + replacement_generation = scheduler._jobs[job_id]["_generation"] + assert replacement_generation > original_generation + assert scheduler._jobs[job_id]["_agent_task_status"] == "running" + + release.set() + + async def wait_until_released() -> None: + while scheduler._handles: + await asyncio.sleep(0) + + await asyncio.wait_for(wait_until_released(), timeout=1) + completed = AgentTaskOper().get(task.id) + assert completed.enabled is False + assert completed.last_status == "success" + assert job_id not in scheduler._jobs + assert scheduler._scheduler.get_job(job_id) is None + + +def test_finished_date_task_cannot_remove_reenabled_job() -> None: + """date 收口后重新启用的任务不再属于旧执行的运行时清理范围。""" + task = _add_agent_task("date", _future_time(), "date-reenabled") + scheduler = _build_agent_task_scheduler() + scheduler.update_agent_task_job(task.id) + job_id = scheduler._get_agent_task_job_id(task.id) + original_generation = scheduler._jobs[job_id]["_generation"] + oper = AgentTaskOper() + run = oper.begin_run(task.id) + assert run is not None + outcome = oper.finish_run_outcome(run.run_id, success=True, result="完成") + assert outcome.date_task_disabled is True + assert oper.update( + task.id, + {"enabled": True, "last_status": "waiting"}, + ) is True + scheduler.update_agent_task_job(task.id) + replacement_generation = scheduler._jobs[job_id]["_generation"] + assert replacement_generation > original_generation + assert scheduler._jobs[job_id]["_agent_task_run_id"] == run.run_id + assert scheduler._jobs[job_id]["_agent_task_status"] == "waiting" + + assert scheduler._remove_agent_task_job_generation( + task.id, + original_generation, + run.run_id, + ) is False + assert scheduler._jobs[job_id]["_generation"] == replacement_generation + assert scheduler._scheduler.get_job(job_id) is not None + + @pytest.mark.parametrize( "run_time_factory", [_past_time, _future_time, _invalid_time], @@ -401,6 +503,7 @@ async def test_interrupted_date_task_manual_run_disables_and_removes_job( manager = SimpleNamespace( execute_scheduled_task=AgentManager.execute_scheduled_task, process_message=process_message, + _accepting_tasks=True, ) manager.execute_scheduled_task = AgentManager.execute_scheduled_task.__get__(manager) monkeypatch.setattr( @@ -674,6 +777,7 @@ async def test_scheduler_config_reload_preserves_active_agent_task( manager = SimpleNamespace( execute_scheduled_task=AgentManager.execute_scheduled_task, process_message=process_message, + _accepting_tasks=True, ) manager.execute_scheduled_task = AgentManager.execute_scheduled_task.__get__(manager) monkeypatch.setattr( diff --git a/tests/test_agent_task_execution_service.py b/tests/test_agent_task_execution_service.py new file mode 100644 index 000000000..0fd72ed49 --- /dev/null +++ b/tests/test_agent_task_execution_service.py @@ -0,0 +1,310 @@ +"""AgentTask 执行服务的取消、终态和调度清理合同。""" + +import asyncio +import threading +from collections.abc import Callable +from uuid import uuid4 + +import pytest + +from app.application.agenttask import AgentTaskExecutionService +from app.db.adapters.transaction import TransactionalWriteRunner +from app.db.oper.agenttask import AgentTaskOper +from app.db.session import SessionFactory, async_session_scope +from app.db.worker import DatabaseWorker +from app.schemas.exception import DatabaseWorkerClosedError + + +def _add_task(prefix: str, *, trigger_type: str = "cron"): + """创建一条与其他用例隔离的可执行任务。""" + 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=None, + source="api", + original_chat_id=None, + ) + + +def _build_service( + worker: DatabaseWorker, + repository: Callable[[object], object] | None = None, +) -> AgentTaskExecutionService: + """按生产事务和 worker 边界构造独立服务。""" + transaction = TransactionalWriteRunner( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + return AgentTaskExecutionService( + repository=repository or (lambda session: AgentTaskOper(session)), + async_executor=worker, + sync_transaction=transaction.sync, + ) + + +async def _wait_for_worker( + worker: DatabaseWorker, + predicate: Callable[[], bool], +) -> None: + """等待 worker 进入目标状态，超时由测试框架明确失败。""" + for _ in range(200): + if predicate(): + return + await asyncio.sleep(0.005) + raise AssertionError(f"数据库 worker 未进入目标状态: {worker.snapshot()}") + + +@pytest.mark.anyio +async def test_cancelled_queued_claim_does_not_create_run() -> None: + """认领尚未开始时取消，应撤销排队工作且不得产生运行记录。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + occupied = threading.Event() + release = threading.Event() + blocker = asyncio.create_task(worker.run( + lambda: (occupied.set(), release.wait()) + )) + await asyncio.to_thread(occupied.wait) + task = _add_task("queued-cancel") + service = _build_service(worker) + + claim = asyncio.create_task(service.claim(task.id)) + await _wait_for_worker(worker, lambda: worker.snapshot().queued == 1) + claim.cancel() + await asyncio.sleep(0) + release.set() + + with pytest.raises(asyncio.CancelledError): + await claim + await blocker + current = AgentTaskOper().get(task.id) + assert current.last_status == "waiting" + assert current.last_run_id is None + assert AgentTaskOper().list_runs(task.id) == [] + await worker.shutdown() + + +@pytest.mark.anyio +async def test_cancelled_started_claim_is_compensated_before_return() -> None: + """认领事务已开始时取消，返回前必须把已提交运行收口为失败。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + started = threading.Event() + release = threading.Event() + + class BlockingRepository: + """在认领已写入但事务尚未提交的位置制造取消窗口。""" + + def __init__(self, session: object) -> None: + self._repository = AgentTaskOper(session) + + def __getattr__(self, name: str): + return getattr(self._repository, name) + + def begin_run(self, *args, **kwargs): + run = self._repository.begin_run(*args, **kwargs) + started.set() + release.wait() + return run + + task = _add_task("started-cancel") + service = _build_service(worker, BlockingRepository) + claim = asyncio.create_task(service.claim(task.id)) + await asyncio.to_thread(started.wait) + claim.cancel() + release.set() + + with pytest.raises(asyncio.CancelledError): + await claim + current = AgentTaskOper().get(task.id) + runs = AgentTaskOper().list_runs(task.id) + assert current.last_status == "failed" + assert current.last_result == "Agent 定时任务已取消" + assert current.run_count == 1 + assert len(runs) == 1 + assert runs[0].status == "failed" + await worker.shutdown() + + +@pytest.mark.anyio +async def test_claim_retries_transient_worker_overload() -> None: + """一次性任务不得因触发瞬间容量已满而永久丢失。""" + worker = DatabaseWorker(max_workers=1, capacity=1) + await worker.start() + task = _add_task("claim-overload", trigger_type="date") + service = _build_service(worker) + + occupied = threading.Event() + release = threading.Event() + blocker = asyncio.create_task(worker.run( + lambda: (occupied.set(), release.wait()) + )) + await asyncio.to_thread(occupied.wait) + claim = asyncio.create_task(service.claim(task.id)) + await _wait_for_worker(worker, lambda: worker.snapshot().rejected > 0) + assert claim.done() is False + release.set() + + claimed = await claim + await blocker + assert claimed.run is not None + current = AgentTaskOper().get(task.id) + assert current.last_status == "running" + assert current.last_run_id == claimed.run.run_id + await service.finalize(claimed.run, success=True, result="完成") + await worker.shutdown() + + +@pytest.mark.anyio +async def test_repeated_finalize_cancellation_waits_for_single_terminal_write() -> None: + """重复取消不得打断已开始的终态事务或重复累计执行次数。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + started = threading.Event() + release = threading.Event() + + class BlockingRepository: + """在运行终态已写入但事务尚未提交的位置制造重复取消窗口。""" + + def __init__(self, session: object) -> None: + self._repository = AgentTaskOper(session) + + def __getattr__(self, name: str): + return getattr(self._repository, name) + + def finish_run_outcome(self, *args, **kwargs): + outcome = self._repository.finish_run_outcome(*args, **kwargs) + started.set() + release.wait() + return outcome + + task = _add_task("finish-cancel") + claim_service = _build_service(worker) + claimed = await claim_service.claim(task.id) + assert claimed.run is not None + service = _build_service(worker, BlockingRepository) + finalize = asyncio.create_task(service.finalize( + claimed.run, + success=True, + result="完成", + )) + await asyncio.to_thread(started.wait) + finalize.cancel() + await asyncio.sleep(0) + finalize.cancel() + release.set() + + with pytest.raises(asyncio.CancelledError): + await finalize + current = AgentTaskOper().get(task.id) + runs = AgentTaskOper().list_runs(task.id) + assert current.last_status == "success" + assert current.run_count == 1 + assert len(runs) == 1 + assert runs[0].status == "success" + await worker.shutdown() + + +@pytest.mark.anyio +async def test_cancelled_queued_finalize_waits_for_terminal_write() -> None: + """终态事务仍在队列时取消，返回前也必须完成唯一一次收口。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + task = _add_task("queued-finish-cancel") + service = _build_service(worker) + claimed = await service.claim(task.id) + assert claimed.run is not None + + occupied = threading.Event() + release = threading.Event() + blocker = asyncio.create_task(worker.run( + lambda: (occupied.set(), release.wait()) + )) + await asyncio.to_thread(occupied.wait) + finalize = asyncio.create_task(service.finalize( + claimed.run, + success=True, + result="完成", + )) + await _wait_for_worker(worker, lambda: worker.snapshot().queued == 1) + finalize.cancel() + release.set() + + with pytest.raises(asyncio.CancelledError): + await finalize + await blocker + current = AgentTaskOper().get(task.id) + runs = AgentTaskOper().list_runs(task.id) + assert current.last_status == "success" + assert current.run_count == 1 + assert len(runs) == 1 + assert runs[0].status == "success" + await worker.shutdown() + + +@pytest.mark.anyio +async def test_finalize_retries_transient_worker_overload() -> None: + """容量暂满时保留终态 owner，取得 admission 后再提交结果。""" + worker = DatabaseWorker(max_workers=1, capacity=1) + await worker.start() + task = _add_task("finish-overload") + service = _build_service(worker) + claimed = await service.claim(task.id) + assert claimed.run is not None + + occupied = threading.Event() + release = threading.Event() + blocker = asyncio.create_task(worker.run( + lambda: (occupied.set(), release.wait()) + )) + await asyncio.to_thread(occupied.wait) + finalize = asyncio.create_task(service.finalize( + claimed.run, + success=True, + result="完成", + )) + await _wait_for_worker(worker, lambda: worker.snapshot().rejected > 0) + assert finalize.done() is False + release.set() + + outcome = await finalize + await blocker + assert outcome.run_finalized is True + assert AgentTaskOper().get(task.id).last_status == "success" + assert worker.snapshot().queued == 0 + assert worker.snapshot().running == 0 + await worker.shutdown() + + +@pytest.mark.anyio +async def test_closed_worker_does_not_finalize_or_remove_schedule() -> None: + """持久化不可用时不得伪造终态或清理运行时调度。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + task = _add_task("closed-finalize", trigger_type="date") + service = _build_service(worker) + claimed = await service.claim(task.id) + assert claimed.run is not None + await worker.shutdown() + removed: list[tuple[int, int]] = [] + + with pytest.raises(DatabaseWorkerClosedError): + await service.finalize( + claimed.run, + success=True, + result="完成", + scheduler_generation=3, + remove_schedule=lambda task_id, generation, _run_id: ( + removed.append((task_id, generation)) or True + ), + ) + + assert removed == [] + assert AgentTaskOper().get(task.id).last_status == "running" diff --git a/tests/test_agent_task_runs.py b/tests/test_agent_task_runs.py index 8cb90a8f3..caf4e9e8c 100644 --- a/tests/test_agent_task_runs.py +++ b/tests/test_agent_task_runs.py @@ -1,8 +1,6 @@ import json from concurrent.futures import ThreadPoolExecutor from threading import Event, Thread, current_thread -from types import SimpleNamespace -from unittest.mock import AsyncMock, Mock from uuid import uuid4 import pytest @@ -17,7 +15,6 @@ from app.db.oper.agenttask import AgentTaskOper from app.db.models.agenttask import AgentTask from app.db.models.agenttaskrun import AgentTaskRun from app.db.session import SessionFactory -from app.scheduler import Scheduler Engine = get_engine() @@ -165,31 +162,6 @@ async def test_agenttask_oper_async_get_uses_async_query_boundary() -> None: assert await AgentTaskOper().async_get(task.id, user_id="another-user") is None -@pytest.mark.anyio -async def test_scheduler_agent_task_cleanup_uses_async_query(monkeypatch) -> None: - """async 调度收尾必须等待异步任务查询，不得退回同步 Oper 调用。""" - execute = AsyncMock(return_value=(True, "执行完成")) - async_get = AsyncMock( - return_value=SimpleNamespace(trigger_type="cron", enabled=True) - ) - sync_get = Mock(side_effect=AssertionError("不应调用同步 AgentTaskOper.get")) - scheduler = SimpleNamespace(remove_agent_task_job=Mock()) - monkeypatch.setattr( - "app.application.agent.get_running_agent_manager", - lambda: SimpleNamespace(execute_scheduled_task=execute), - ) - monkeypatch.setattr(AgentTaskOper, "async_get", async_get) - monkeypatch.setattr(AgentTaskOper, "get", sync_get) - - result = await Scheduler.execute_agent_task(scheduler, task_id=42) - - assert result == (True, "执行完成") - execute.assert_awaited_once_with(42, trigger_source="scheduled") - async_get.assert_awaited_once_with(42) - sync_get.assert_not_called() - scheduler.remove_agent_task_job.assert_not_called() - - def test_begin_run_rolls_back_task_claim_when_run_insert_fails() -> None: """运行记录插入失败时，任务的 running 投影必须随事务回滚。""" first_task = _add_task("run-rollback-first") @@ -258,7 +230,10 @@ def test_stale_finish_cannot_overwrite_latest_run_projection() -> None: second = oper.begin_run(task.id, "manual") assert second - assert oper.finish_run(first.run_id, success=True, result="旧结果") + outcome = oper.finish_run_outcome(first.run_id, success=True, result="旧结果") + assert outcome.run_finalized is True + assert outcome.task_projection_updated is False + assert outcome.date_task_disabled is False current = oper.get(task.id) assert current.last_run_id == second.run_id assert current.last_status == "running" diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 36b2941cc..b9c389cee 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -599,6 +599,30 @@ def test_agent_consumers_use_explicit_data_port_getters(): assert violations == [] +def test_scheduler_does_not_depend_on_database_implementation(): + """Scheduler 只能消费应用端口，不得重新直连 app.db 实现。""" + dependencies = _build_module_graph().get("app.scheduler", set()) + assert { + dependency + for dependency in dependencies + if dependency == "app.db" or dependency.startswith("app.db.") + } == set() + + +def test_agent_task_async_execution_uses_application_service(): + """AgentTask async 执行不得经动态数据端口隐藏同步 Oper 调用。""" + path = APP_ROOT / "agent" / "orchestrator.py" + tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + violations = [ + node.lineno + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) + and node.module == "app.application.agentdata" + and any(alias.name == "get_agent_task_port" for alias in node.names) + ] + assert violations == [] + + def test_monitor_dispatcher_uses_explicit_history_port_getter(): """监控分发器不得把兼容 TransferHistoryPort 伪装成数据库 Oper。""" path = APP_ROOT / "monitor" / "dispatcher.py"