diff --git a/app/db/oper/agenttask.py b/app/db/oper/agenttask.py index f9ad2fa7b..cfc5a8861 100644 --- a/app/db/oper/agenttask.py +++ b/app/db/oper/agenttask.py @@ -4,6 +4,7 @@ from datetime import datetime from typing import Optional from uuid import uuid4 +from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from app.db.base import DbOper @@ -66,6 +67,25 @@ class AgentTaskOper(DbOper): return query(self._db) return run_sync_transaction(query) + async def async_get( + self, + task_id: int, + user_id: Optional[str] = None, + ) -> Optional[AgentTask]: + """通过异步会话查询单个 Agent 定时任务。""" + async def query(session: AsyncSession) -> Optional[AgentTask]: + """在调用方异步会话中执行与同步入口相同的查询语义。""" + result = await session.execute( + _get_for_user_statement( + AgentTask, + task_id=task_id, + user_id=user_id, + ) + ) + return result.scalars().first() + + return await self._execute_async_query(query) + def list( self, user_id: Optional[str] = None, diff --git a/app/scheduler.py b/app/scheduler.py index 1128de57c..2a5b3720e 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -1600,7 +1600,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): trigger_source=trigger_source, ) finally: - task = AgentTaskOper().get(task_id) + 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) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index c15a0783d..f4fd0093b 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -19,7 +19,8 @@ - 启动性能探针会在隔离生命周期中真实创建并释放 TaskRegistry;normal/safe 组件数分别为 `23`/`11`,CI 只读检查使用稳定的宿主模块集合和生命周期组件顺序,不再把 Python/平台模块数量当作硬合同。 - 官方插件快照覆盖 `plugins.v3`、`plugins.v2` 以及 V3 实际会从 `package.json` 回退加载的 31 个默认实现;`app/plugins/**` 仍只是宿主运行副本,不进入扫描。 - SDK 快照以各模块显式 `__all__` 为公开合同,能够记录赋值别名;`typing`、`__future__` 等实现期导入不再被误冻结,既有数据库备份门面已补精确导出清单。 -- async 阻塞实际债务已由 fixture 中的 10 项下降到 1 项并固化低水位;剩余项是 Scheduler Agent task 查询,后续阶段迁入异步查询边界后归零。 +- async 阻塞实际债务已由 fixture 中的 10 项归零;Scheduler Agent task 收尾查询已复用 + `AgentTaskOper.async_get` 的统一 AsyncSession 边界,CI 继续以零债务基线拒绝回退。 本阶段只修复治理信号和事实源,不把基线刷新当作业务重构完成。后台任务所有权、Module Contract V2、typed runtime、durable 副作用和质量规模化仍按下列 P1/P2 顺序推进。 diff --git a/tests/fixtures/architecture/async-blocking-baseline.json b/tests/fixtures/architecture/async-blocking-baseline.json index 4fcc9c251..0967ef424 100644 --- a/tests/fixtures/architecture/async-blocking-baseline.json +++ b/tests/fixtures/architecture/async-blocking-baseline.json @@ -1,3 +1 @@ -{ - "app/scheduler.py:Scheduler.execute_agent_task:AgentTaskOper.get": 1 -} +{} diff --git a/tests/test_agent_task_runs.py b/tests/test_agent_task_runs.py index 3c0aa3e66..eebbb4949 100644 --- a/tests/test_agent_task_runs.py +++ b/tests/test_agent_task_runs.py @@ -1,6 +1,8 @@ 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 @@ -15,6 +17,7 @@ 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() @@ -153,6 +156,40 @@ def test_agenttaskrun_oper_reuses_explicit_query_session(db, monkeypatch): assert oper.list_runs(task.id) +@pytest.mark.anyio +async def test_agenttask_oper_async_get_uses_async_query_boundary() -> None: + """异步任务查询应复用统一 AsyncSession 路径并保持 owner 过滤语义。""" + task = _add_task("run-async-query") + + assert await AgentTaskOper().async_get(task.id, user_id=task.user_id) is not 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.agent.runtime_loader.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")