From 2031f3c4207a6e6a26ab40845d7312f239b93ffb Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 23 Aug 2026 00:32:04 +0800 Subject: [PATCH] refactor: migrate agent task query ownership --- app/db/models/agenttask.py | 43 +++++++++++++------ app/db/oper/agenttask.py | 37 ++++++++++++++-- .../backend-architecture-next-stage.md | 10 +++-- docs/rules/10-data-and-persistent.md | 2 +- .../transaction-debt-baseline.json | 14 +----- tests/test_db_plugin_message_agent_queries.py | 18 ++++++++ 6 files changed, 92 insertions(+), 32 deletions(-) diff --git a/app/db/models/agenttask.py b/app/db/models/agenttask.py index 0bc1d8f82..920f1a1d2 100644 --- a/app/db/models/agenttask.py +++ b/app/db/models/agenttask.py @@ -4,7 +4,32 @@ from sqlalchemy import Boolean, Index, Integer, String, Text, select, update from sqlalchemy.orm import Mapped, Session, mapped_column from app.db.base import Base, execute_dml, get_id_column -from app.db.decorators import db_query + + +def _get_for_user_statement( + model: type["AgentTask"], + task_id: int, + user_id: Optional[str] = None, +): + """构造按任务 ID 与可选用户归属收窄的查询语句。""" + statement = select(model).where(model.id == task_id) + if user_id is not None: + statement = statement.where(model.user_id == user_id) + return statement + + +def _list_for_user_statement( + model: type["AgentTask"], + user_id: Optional[str] = None, + enabled: Optional[bool] = None, +): + """构造按用户、启用状态和创建时间排序的任务列表语句。""" + statement = select(model) + if user_id is not None: + statement = statement.where(model.user_id == user_id) + if enabled is not None: + statement = statement.where(model.enabled.is_(enabled)) + return statement.order_by(model.created_at.desc(), model.id.desc()) class AgentTask(Base): @@ -59,7 +84,6 @@ class AgentTask(Base): return task.id @classmethod - @db_query def get_for_user( cls, db: Session, @@ -69,13 +93,11 @@ class AgentTask(Base): """ 按任务 ID 和可选用户 ID 查询 Agent 定时任务。 """ - statement = select(cls).where(cls.id == task_id) - if user_id is not None: - statement = statement.where(cls.user_id == user_id) - return db.execute(statement).scalars().first() + return db.execute( + _get_for_user_statement(cls, task_id=task_id, user_id=user_id) + ).scalars().first() @classmethod - @db_query def list_for_user( cls, db: Session, @@ -85,13 +107,8 @@ class AgentTask(Base): """ 按用户和启用状态查询 Agent 定时任务。 """ - statement = select(cls) - if user_id is not None: - statement = statement.where(cls.user_id == user_id) - if enabled is not None: - statement = statement.where(cls.enabled.is_(enabled)) return list(db.execute( - statement.order_by(cls.created_at.desc(), cls.id.desc()) + _list_for_user_statement(cls, user_id=user_id, enabled=enabled) ).scalars().all()) @classmethod diff --git a/app/db/oper/agenttask.py b/app/db/oper/agenttask.py index 4ce59eea7..ccec6a81e 100644 --- a/app/db/oper/agenttask.py +++ b/app/db/oper/agenttask.py @@ -4,9 +4,16 @@ from datetime import datetime from typing import Optional from uuid import uuid4 +from sqlalchemy.orm import Session + from app.db.base import DbOper -from app.db.models.agenttask import AgentTask +from app.db.models.agenttask import ( + AgentTask, + _get_for_user_statement, + _list_for_user_statement, +) from app.db.models.agenttaskrun import AgentTaskRun +from app.db.uow import run_sync_transaction class AgentTaskOper(DbOper): @@ -45,7 +52,19 @@ class AgentTaskOper(DbOper): """ 查询单个 Agent 定时任务。 """ - return AgentTask.get_for_user(self._db, task_id=task_id, user_id=user_id) + def query(session: Session) -> Optional[AgentTask]: + """在调用方会话中读取单个任务。""" + return session.execute( + _get_for_user_statement( + AgentTask, + task_id=task_id, + user_id=user_id, + ) + ).scalars().first() + + if isinstance(self._db, Session): + return query(self._db) + return run_sync_transaction(query) def list( self, @@ -55,7 +74,19 @@ class AgentTaskOper(DbOper): """ 查询 Agent 定时任务列表。 """ - return AgentTask.list_for_user(self._db, user_id=user_id, enabled=enabled) + def query(session: Session) -> list[AgentTask]: + """在调用方会话中读取任务列表。""" + return list(session.execute( + _list_for_user_statement( + AgentTask, + user_id=user_id, + enabled=enabled, + ) + ).scalars().all()) + + if isinstance(self._db, Session): + return query(self._db) + return run_sync_transaction(query) def update( self, diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 6a9c49632..381b7dce9 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -426,8 +426,12 @@ flowchart TB 回调交给 Command；commit/flush 失败回滚，事件或上报失败只传播原异常，不回滚已提交记录。 - 同步/异步 `SubscribeChain.add` 方法长度从各 203 行降至 183/186 行；新增 9 个事务边界测试， 覆盖成功顺序、commit/flush 失败、重复请求、Oper 不提交、事件失败、上报失败与真实落库。 -- Model 装饰器总数仍为 178：本切片绕开了继承自 `Base.create/async_create` 的自动提交， - 但为保留既有 Model/旧 SDK 查询兼容未机械删除查询装饰器；ratchet 保持不增，后续切片继续下降。 +- Model 查询装饰器此前为 123 个：本切片绕开了继承自 `Base.create/async_create` 的自动提交， + 并继续保留既有 Model/旧 SDK 查询兼容；本次 AgentTask 切片将查询装饰器减少到 121 个。 + 2026-08-23 已完成 AgentTask 查询切片：`AgentTaskOper.get/list` 直接在调用方 Session 中执行查询， + `AgentTask.get_for_user/list_for_user` 保留原签名和返回语义供旧调用方使用，但不再持有查询装饰器； + 无 Session 的旧 Oper 入口继续由组合根兼容事务执行器承接。查询装饰器低水位由 123 降至 121， + 归属过滤、启用状态过滤和创建时间/主键稳定排序由 canonical Oper 测试覆盖。 #### ARCH-222：按风险迁移其余写用例 @@ -1191,7 +1195,7 @@ rollback: | 基线写入行为 | 默认命令可能覆盖 fixture | 所有默认/check 命令保证工作树不变；write 必须显式 scope | | 全功能 worker | 配置允许 >1，控制面会复制 | 启动期明确拒绝 >1；文档与配置一致 | | 健康接口 | 认证 `/system/ping` 为主 | 分离公开 live 与受限/安全 ready；失败原因可诊断 | -| Model 事务装饰器 | 当前 123 个且全部只读；写装饰器 0 | 查询债务只降不增；写事务不回退到 Model/Base 隐式提交 | +| Model 事务装饰器 | 当前 121 个且全部只读；写装饰器 0 | 查询债务只降不增；写事务不回退到 Model/Base 隐式提交 | | 新写用例事务 | 宿主写 Oper 已脱离 Base 隐式提交 | 100% 由入口/Application 边界拥有 Session/UoW | | 高频 Module 契约 | 212 个宿主能力显式登记 | 新观察到的宿主方法必须同步登记完整契约 | | Event payload | 53 类型全部登记 typed payload 与可靠性 | 新事件必须同步登记，不回退裸 dict | diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index a4d8caae2..5dff43169 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -84,7 +84,7 @@ Oper classes accept and return persistence values. Turning a `MediaInfo` or ### Transaction ownership ratchet - `tests/fixtures/architecture/transaction-debt-baseline.json` records the - existing Model transaction decorators. The current 123 decorators are query-only + existing Model transaction decorators. The current 121 decorators are query-only migration debt: they may decrease but must never increase or move to a new Model method. Both `db_update` and `async_db_update` must remain at zero. - New Model methods must not use `db_query`, `db_update`, `async_db_query`, or diff --git a/tests/fixtures/architecture/transaction-debt-baseline.json b/tests/fixtures/architecture/transaction-debt-baseline.json index 71c7ac16d..5e1a68cee 100644 --- a/tests/fixtures/architecture/transaction-debt-baseline.json +++ b/tests/fixtures/architecture/transaction-debt-baseline.json @@ -3,10 +3,10 @@ "by_kind": { "async_db_query": 49, "async_db_update": 0, - "db_query": 74, + "db_query": 72, "db_update": 0 }, - "count": 123, + "count": 121, "methods": [ { "decorator": "async_db_query", @@ -28,16 +28,6 @@ "file": "app/db/models/agentchat.py", "method": "AgentChat.list_by_page" }, - { - "decorator": "db_query", - "file": "app/db/models/agenttask.py", - "method": "AgentTask.get_for_user" - }, - { - "decorator": "db_query", - "file": "app/db/models/agenttask.py", - "method": "AgentTask.list_for_user" - }, { "decorator": "db_query", "file": "app/db/models/agenttaskrun.py", diff --git a/tests/test_db_plugin_message_agent_queries.py b/tests/test_db_plugin_message_agent_queries.py index 2579784e4..43e399cdc 100644 --- a/tests/test_db_plugin_message_agent_queries.py +++ b/tests/test_db_plugin_message_agent_queries.py @@ -13,6 +13,7 @@ from app.db.models.agenttask import AgentTask from app.db.models.downloadfailure import DownloadFailure from app.db.models.message import Message from app.db.models.plugindata import PluginData +from app.db.oper.agenttask import AgentTaskOper @pytest.fixture(autouse=True) @@ -307,6 +308,23 @@ def test_agenttask_get_for_user_enforces_ownership(db): assert AgentTask.get_for_user(db.session, task_id, user_id="bob") is None +def test_agenttask_oper_reads_with_explicit_session(db, monkeypatch): + """AgentTaskOper 的宿主查询使用调用方 Session，不再经过旧事务兼容执行器。""" + task_id = AgentTask.add_task(db.session, **_task("canonical", user_id="alice")) + + monkeypatch.setattr( + "app.db.oper.agenttask.run_sync_transaction", + lambda _query: pytest.fail("显式 Session 查询不应创建兼容事务"), + ) + + oper = AgentTaskOper(db.session) + task = oper.get(task_id, user_id="alice") + tasks = oper.list(user_id="alice", enabled=True) + + assert task is not None and task.id == task_id + assert [item.id for item in tasks] == [task_id] + + def test_agenttask_list_for_user_filters_by_owner_and_enabled(db): """ 列表按归属与启用状态收窄，并按创建时间倒序。