refactor: remove final async blocking query

This commit is contained in:
jxxghp
2026-08-23 23:46:15 +08:00
parent ba2eb869e2
commit 2372b1236e
5 changed files with 61 additions and 5 deletions
+20
View File
@@ -4,6 +4,7 @@ from datetime import datetime
from typing import Optional from typing import Optional
from uuid import uuid4 from uuid import uuid4
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.db.base import DbOper from app.db.base import DbOper
@@ -66,6 +67,25 @@ class AgentTaskOper(DbOper):
return query(self._db) return query(self._db)
return run_sync_transaction(query) 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( def list(
self, self,
user_id: Optional[str] = None, user_id: Optional[str] = None,
+1 -1
View File
@@ -1600,7 +1600,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
trigger_source=trigger_source, trigger_source=trigger_source,
) )
finally: finally:
task = AgentTaskOper().get(task_id) task = await AgentTaskOper().async_get(task_id)
if task and task.trigger_type == "date" and not task.enabled: if task and task.trigger_type == "date" and not task.enabled:
self.remove_agent_task_job(task_id) self.remove_agent_task_job(task_id)
@@ -19,7 +19,8 @@
- 启动性能探针会在隔离生命周期中真实创建并释放 TaskRegistrynormal/safe 组件数分别为 `23`/`11`,CI 只读检查使用稳定的宿主模块集合和生命周期组件顺序,不再把 Python/平台模块数量当作硬合同。 - 启动性能探针会在隔离生命周期中真实创建并释放 TaskRegistrynormal/safe 组件数分别为 `23`/`11`,CI 只读检查使用稳定的宿主模块集合和生命周期组件顺序,不再把 Python/平台模块数量当作硬合同。
- 官方插件快照覆盖 `plugins.v3``plugins.v2` 以及 V3 实际会从 `package.json` 回退加载的 31 个默认实现;`app/plugins/**` 仍只是宿主运行副本,不进入扫描。 - 官方插件快照覆盖 `plugins.v3``plugins.v2` 以及 V3 实际会从 `package.json` 回退加载的 31 个默认实现;`app/plugins/**` 仍只是宿主运行副本,不进入扫描。
- SDK 快照以各模块显式 `__all__` 为公开合同,能够记录赋值别名;`typing``__future__` 等实现期导入不再被误冻结,既有数据库备份门面已补精确导出清单。 - 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 顺序推进。 本阶段只修复治理信号和事实源,不把基线刷新当作业务重构完成。后台任务所有权、Module Contract V2、typed runtime、durable 副作用和质量规模化仍按下列 P1/P2 顺序推进。
+1 -3
View File
@@ -1,3 +1 @@
{ {}
"app/scheduler.py:Scheduler.execute_agent_task:AgentTaskOper.get": 1
}
+37
View File
@@ -1,6 +1,8 @@
import json import json
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from threading import Event, Thread, current_thread from threading import Event, Thread, current_thread
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
from uuid import uuid4 from uuid import uuid4
import pytest import pytest
@@ -15,6 +17,7 @@ from app.db.oper.agenttask import AgentTaskOper
from app.db.models.agenttask import AgentTask from app.db.models.agenttask import AgentTask
from app.db.models.agenttaskrun import AgentTaskRun from app.db.models.agenttaskrun import AgentTaskRun
from app.db.session import SessionFactory from app.db.session import SessionFactory
from app.scheduler import Scheduler
Engine = get_engine() Engine = get_engine()
@@ -153,6 +156,40 @@ def test_agenttaskrun_oper_reuses_explicit_query_session(db, monkeypatch):
assert oper.list_runs(task.id) 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: def test_begin_run_rolls_back_task_claim_when_run_insert_fails() -> None:
"""运行记录插入失败时,任务的 running 投影必须随事务回滚。""" """运行记录插入失败时,任务的 running 投影必须随事务回滚。"""
first_task = _add_task("run-rollback-first") first_task = _add_task("run-rollback-first")