diff --git a/app/db/models/agentchat.py b/app/db/models/agentchat.py index 4fc948997..1537d5a9b 100644 --- a/app/db/models/agentchat.py +++ b/app/db/models/agentchat.py @@ -5,7 +5,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, Session, mapped_column from app.db.base import Base, get_id_column -from app.db.decorators import async_db_query, db_query +from app.db.decorators import legacy_async_db_query, legacy_db_query class AgentChat(Base): @@ -50,7 +50,7 @@ class AgentChat(Base): ) @classmethod - @db_query + @legacy_db_query def get_by_session( cls, db: Session, session_id: str, user_id: Optional[str] = None ) -> Optional["AgentChat"]: @@ -63,7 +63,7 @@ class AgentChat(Base): return db.execute(statement.order_by(cls.id.desc())).scalars().first() @classmethod - @async_db_query + @legacy_async_db_query async def async_get_by_session( cls, db: AsyncSession, session_id: str, user_id: Optional[str] = None ) -> Optional["AgentChat"]: @@ -77,7 +77,7 @@ class AgentChat(Base): return result.scalars().first() @classmethod - @db_query + @legacy_db_query def list_by_page( cls, db: Session, @@ -103,7 +103,7 @@ class AgentChat(Base): ).scalars().all()) @classmethod - @async_db_query + @legacy_async_db_query async def async_list_by_page( cls, db: AsyncSession, diff --git a/app/db/models/agenttaskrun.py b/app/db/models/agenttaskrun.py index 8c37607db..1b4d7fb12 100644 --- a/app/db/models/agenttaskrun.py +++ b/app/db/models/agenttaskrun.py @@ -4,7 +4,7 @@ from sqlalchemy import Index, Integer, String, Text, delete, 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 +from app.db.decorators import legacy_db_query from app.db.models.agenttask import AgentTask @@ -249,7 +249,7 @@ class AgentTaskRun(Base): return True @classmethod - @db_query + @legacy_db_query def get_by_run_id( cls, db: Session, @@ -261,7 +261,7 @@ class AgentTaskRun(Base): ).scalars().first() @classmethod - @db_query + @legacy_db_query def list_for_task( cls, db: Session, diff --git a/app/db/oper/agentchat.py b/app/db/oper/agentchat.py index b81d7b612..66c9240d5 100644 --- a/app/db/oper/agentchat.py +++ b/app/db/oper/agentchat.py @@ -77,7 +77,9 @@ class AgentChatOper(DbOper): """ 获取 Agent 会话。 """ - return AgentChat.get_by_session(self._db, session_id, user_id) + return self._execute_sync_query( + lambda session: AgentChat.get_by_session(session, session_id, user_id) + ) async def async_get( self, session_id: str, user_id: Optional[str] = None @@ -85,7 +87,9 @@ class AgentChatOper(DbOper): """ 异步获取 Agent 会话。 """ - return await AgentChat.async_get_by_session(self._db, session_id, user_id) + return await self._execute_async_query( + lambda session: AgentChat.async_get_by_session(session, session_id, user_id) + ) def ensure_session( self, @@ -295,12 +299,14 @@ class AgentChatOper(DbOper): """ 异步分页获取 Agent 会话历史。 """ - return await AgentChat.async_list_by_page( - self._db, - page=page, - count=count, - user_id=user_id, - username=username, + return await self._execute_async_query( + lambda session: AgentChat.async_list_by_page( + session, + page=page, + count=count, + user_id=user_id, + username=username, + ) ) async def async_delete( diff --git a/app/db/oper/agenttask.py b/app/db/oper/agenttask.py index ccec6a81e..f9ad2fa7b 100644 --- a/app/db/oper/agenttask.py +++ b/app/db/oper/agenttask.py @@ -182,7 +182,9 @@ class AgentTaskOper(DbOper): def get_run(self, run_id: str) -> Optional[AgentTaskRun]: """查询一次 Agent 任务运行。""" - return AgentTaskRun.get_by_run_id(self._db, run_id=run_id) + return self._execute_sync_query( + lambda session: AgentTaskRun.get_by_run_id(session, run_id=run_id) + ) def list_runs( self, @@ -191,11 +193,13 @@ class AgentTaskOper(DbOper): limit: int = 10, ) -> list[AgentTaskRun]: """查询任务最近的有界运行历史。""" - return AgentTaskRun.list_for_task( - self._db, - task_id=task_id, - user_id=user_id, - limit=limit, + return self._execute_sync_query( + lambda session: AgentTaskRun.list_for_task( + session, + task_id=task_id, + user_id=user_id, + limit=limit, + ) ) def finish_run( diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 72df34a93..59b8d151c 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -378,7 +378,7 @@ flowchart LR 成功后执行。订阅新增样板由 `startup/subscription.py` 创建独占 Session， `application/subscription/write.py` 决定事务与 post-commit 边界，`SubscribeOper.stage_add()` 只查重、`add` 和 `flush`。旧 SDK 显式构造的无会话 Oper 暂留兼容自动短会话，不得被新代码复用。 - `transaction-debt-baseline.json` 当前冻结 18 个正式只读查询装饰器；原有同步/异步写装饰器 + `transaction-debt-baseline.json` 当前冻结 12 个正式只读查询装饰器；原有同步/异步写装饰器 已全部移除，`db_update` 与 `async_db_update` 必须持续保持为 0。下载/整理历史的旧插件 Model 与工作流、媒体服务器、站点用户数据旧插件 Model 调用由 `legacy_*` 兼容外壳承接，宿主 Oper 必须显式传递 Session。宿主 Oper 也不得调用 Base 保留的 `create/update/delete/truncate` 兼容包装器；AST 门禁保证显式 Session 的提交权不会被底层抢走。 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 636fd226c..f8fcea981 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -28,7 +28,7 @@ 1. **后台任务的统一所有权已覆盖 API 入口，但仍有更深层任务机制待分级。** `app/runtime/tasks.py` 已建立 lifespan 级 TaskRegistry，启动收尾、插件 Release 刷新、Webhook E0 广播、CookieCloud E1 手工调度、消息入口、Seerr 订阅和 WebAgent 断线后执行/快照保存均不再维护端点模块级任务集合或 Starlette 回调，shutdown 会停止接收、取消并有限等待，且生命周期清单明确登记其顺序。主仓 `app/` 已无裸 FastAPI `BackgroundTasks`；当前仍有约 `50` 个更底层 `create_task`/等价任务创建点，与线程池和 APScheduler 并存，后续需逐项确认 owner、取消、等待、重试、幂等和是否 durable，关键业务副作用优先接入已有 Outbox/恢复表。 2. **动态模块契约仍以 legacy 聚合语义为主。** 当前登记 `212` 个模块方法，其中 `194` 个仍使用 `legacy` aggregation，只有 `14` 个 `first_non_empty`、`4` 个 `ordered_list_merge`。`app/runtime/extensions/module/contracts.py:422-455` 已能登记 family、输入/结果标签和基础签名诊断，但 `193` 个方法没有 required parameters，调度器 `app/runtime/extensions/module/dispatcher.py:109-260` 仍主要依赖运行时反射、返回值形状和短路规则。未知第三方方法保留 legacy fallback 是兼容要求，不应删除；宿主高频能力则应逐族补齐可执行的输入校验、结果校验、超时和错误语义。 -3. **查询侧数据库兼容 ABI 仍未完全收口。** 写事务装饰器已降为 `0`，正式 `db_query/async_db_query` 已降至 `18` 个（`9` 个同步、`9` 个异步）。站点、消息、用户、订阅、下载/整理历史、工作流、MediaServer 和 SiteUserData 的宿主 Oper 已迁到显式 Session 路径；对应旧插件 Model 调用由独立 `legacy_*` 外壳保留，可同时接受显式 Session 与无 Session 的位置/关键字参数。剩余正式装饰器仍会隐式创建会话，查询返回的 ORM 对象也可能跨层流转，后续继续按 AgentChat、AgentTaskRun、TransferPending、SystemConfig、PassKey 和 SubscribeHistory 等风险切片迁移。 +3. **查询侧数据库兼容 ABI 仍未完全收口。** 写事务装饰器已降为 `0`，正式 `db_query/async_db_query` 已降至 `12` 个（`5` 个同步、`7` 个异步）。站点、消息、用户、订阅、下载/整理历史、工作流、MediaServer、SiteUserData、AgentChat 和 AgentTaskRun 的宿主 Oper 已迁到显式 Session 路径；对应旧插件 Model 调用由独立 `legacy_*` 外壳保留，可同时接受显式 Session 与无 Session 的位置/关键字参数。剩余正式装饰器仍会隐式创建会话，查询返回的 ORM 对象也可能跨层流转，后续继续按 TransferPending、SystemConfig、PassKey 和 SubscribeHistory 等风险切片迁移。 4. **组合根和全局状态仍形成复杂的隐式运行时图。** Singleton 实例、模块级 provider、`configure_*` 注册函数和兼容 Facade 同时存在；它们解决了旧 ABI 和启动顺序问题，但增加测试污染、重复装配、实例身份和初始化顺序风险。`app/startup/lifecycle/__init__.py:161-376` 已有声明式生命周期，`app/startup/modules_initializer.py:505-530` 也有分阶段关闭，但尚未做到所有进程级资源都只通过 typed HostRuntime 访问。后续应以“新代码禁止新增 Service Locator/Singleton 依赖、旧入口有命中观测”为 ratchet。 ### P2：中长期可演进性债务 @@ -1194,6 +1194,12 @@ host/plugin 架构基线均通过。 Session 不创建额外会话，无 Session 的位置参数和关键字参数继续由 `legacy_*` 外壳兼容；专项测试 `158 passed`，四分片全量测试 `5539 passed, 3 skipped`，host/plugin 架构基线和 Pylint 均通过。 +2026-08-23 完成 AgentChat 与 AgentTaskRun 查询切片：`AgentChatOper`、`AgentTaskOper` 的查询入口 +统一通过 `_execute_sync_query` / `_execute_async_query` 复用调用方 Session，正式查询装饰器由 18 +降至 12 个（同步 5、异步 7），写装饰器保持 0。旧插件仍可直接调用对应 Model 方法，显式 Session +不创建额外会话，无 Session 的关键字调用继续由 `legacy_*` 外壳兼容；专项测试 `44 passed`，四分片 +全量测试 `5543 passed, 4 skipped`。 + #### ARCH-272：异步阻塞检测 **目标**：对新 API/Agent/Application async 路径检测 `open`、文件遍历、同步 HTTP、阻塞 sleep 和重 CPU 解析。 @@ -1371,7 +1377,7 @@ rollback: | 基线写入行为 | 默认命令可能覆盖 fixture | 所有默认/check 命令保证工作树不变；write 必须显式 scope | | 全功能 worker | 配置允许 >1，控制面会复制 | 启动期明确拒绝 >1；文档与配置一致 | | 健康接口 | 认证 `/system/ping` 为主 | 分离公开 live 与受限/安全 ready；失败原因可诊断 | -| Model 事务装饰器 | 当前 18 个且全部只读；写装饰器 0 | 查询债务只降不增；写事务不回退到 Model/Base 隐式提交 | +| Model 事务装饰器 | 当前 12 个且全部只读；写装饰器 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 aa63b39ba..8990fa6e6 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 18 decorators are query-only + existing Model transaction decorators. The current 12 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. - `legacy_db_query` / `legacy_async_db_query` are compatibility-only shells for diff --git a/tests/fixtures/architecture/transaction-debt-baseline.json b/tests/fixtures/architecture/transaction-debt-baseline.json index 1353d6aa4..86db733ab 100644 --- a/tests/fixtures/architecture/transaction-debt-baseline.json +++ b/tests/fixtures/architecture/transaction-debt-baseline.json @@ -1,43 +1,13 @@ { "model_decorators": { "by_kind": { - "async_db_query": 9, + "async_db_query": 7, "async_db_update": 0, - "db_query": 9, + "db_query": 5, "db_update": 0 }, - "count": 18, + "count": 12, "methods": [ - { - "decorator": "async_db_query", - "file": "app/db/models/agentchat.py", - "method": "AgentChat.async_get_by_session" - }, - { - "decorator": "async_db_query", - "file": "app/db/models/agentchat.py", - "method": "AgentChat.async_list_by_page" - }, - { - "decorator": "db_query", - "file": "app/db/models/agentchat.py", - "method": "AgentChat.get_by_session" - }, - { - "decorator": "db_query", - "file": "app/db/models/agentchat.py", - "method": "AgentChat.list_by_page" - }, - { - "decorator": "db_query", - "file": "app/db/models/agenttaskrun.py", - "method": "AgentTaskRun.get_by_run_id" - }, - { - "decorator": "db_query", - "file": "app/db/models/agenttaskrun.py", - "method": "AgentTaskRun.list_for_task" - }, { "decorator": "async_db_query", "file": "app/db/models/passkey.py", diff --git a/tests/test_agent_task_runs.py b/tests/test_agent_task_runs.py index abb6190d5..5d80d91bc 100644 --- a/tests/test_agent_task_runs.py +++ b/tests/test_agent_task_runs.py @@ -14,6 +14,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.db import decorators Engine = get_engine() @@ -134,6 +135,35 @@ def test_begin_run_rejects_unknown_trigger_source() -> None: assert AgentTaskOper().list_runs(task.id) == [] +def test_agenttaskrun_oper_reuses_explicit_query_session(db, monkeypatch): + """AgentTaskOper 的运行记录查询必须复用调用方同步会话。""" + task = _add_task("run-explicit-query") + run = AgentTaskOper().begin_run(task.id) + assert run + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (_ for _ in ()).throw(AssertionError("不应创建额外同步会话")), + ) + + oper = AgentTaskOper(db.session) + assert oper.get_run(run.run_id) is not None + assert oper.list_runs(task.id) + + +def test_agenttaskrun_model_legacy_query_keeps_keyword_abi(monkeypatch): + """旧插件以关键字直调 AgentTaskRun 时仍自动补入短会话。""" + opened = [] + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (opened.append(True) or SessionFactory()), + ) + + assert AgentTaskRun.get_by_run_id(run_id="missing-legacy") is None + assert opened == [True] + + def test_begin_run_rolls_back_task_claim_when_run_insert_fails() -> None: """运行记录插入失败时，任务的 running 投影必须随事务回滚。""" first_task = _add_task("run-rollback-first") diff --git a/tests/test_architecture_contract_baseline.py b/tests/test_architecture_contract_baseline.py index 388322288..d2e1778ad 100644 --- a/tests/test_architecture_contract_baseline.py +++ b/tests/test_architecture_contract_baseline.py @@ -126,8 +126,8 @@ def test_transaction_debt_baseline_is_a_model_and_oper_ratchet() -> None: baseline = json.loads(baseline_path.read_text(encoding="utf-8")) assert baseline["schema_version"] == 1 - assert baseline["model_decorators"]["count"] == 18 - assert sum(baseline["model_decorators"]["by_kind"].values()) == 18 + assert baseline["model_decorators"]["count"] == 12 + assert sum(baseline["model_decorators"]["by_kind"].values()) == 12 assert baseline["model_decorators"]["by_kind"]["db_update"] == 0 assert baseline["model_decorators"]["by_kind"]["async_db_update"] == 0 assert baseline["model_transaction_calls"] == {"count": 0, "calls": []} diff --git a/tests/test_db_plugin_message_agent_queries.py b/tests/test_db_plugin_message_agent_queries.py index fd8ade0a1..9e60c9e98 100644 --- a/tests/test_db_plugin_message_agent_queries.py +++ b/tests/test_db_plugin_message_agent_queries.py @@ -8,12 +8,14 @@ import asyncio import pytest +from app.db import decorators from app.db.models.agentchat import AgentChat 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 +from app.db.session import SessionFactory @pytest.fixture(autouse=True) @@ -238,6 +240,48 @@ def test_agentchat_get_by_session_enforces_user_scope(db): assert asyncio.run(AgentChat.async_get_by_session(session_id="s-owned", user_id="bob")) is None +def test_agentchat_oper_reuses_explicit_query_sessions(db, monkeypatch): + """AgentChatOper 的同步与异步查询必须复用调用方会话。""" + db.add(_chat("s-explicit", user_id="explicit")) + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (_ for _ in ()).throw(AssertionError("不应创建额外同步会话")), + ) + + from app.db.oper.agentchat import AgentChatOper + + assert AgentChatOper(db.session).get("s-explicit", "explicit") is not None + + async def check() -> None: + """验证异步 Agent 会话查询复用显式 AsyncSession。""" + from app.db.session import async_session_scope + + async with async_session_scope() as session: + monkeypatch.setattr( + decorators, + "async_session_scope", + lambda: (_ for _ in ()).throw(AssertionError("不应创建额外异步会话")), + ) + assert await AgentChatOper(session).async_get("s-explicit", "explicit") + + asyncio.run(check()) + + +def test_agentchat_model_legacy_query_keeps_keyword_abi(db, monkeypatch): + """旧插件以关键字直调 AgentChat 时仍自动补入短会话。""" + db.add(_chat("s-legacy")) + opened = [] + monkeypatch.setattr( + decorators, + "ScopedSession", + lambda: (opened.append(True) or SessionFactory()), + ) + + assert AgentChat.get_by_session(session_id="s-legacy") is not None + assert opened == [True] + + def test_agentchat_list_by_page_matches_either_user_or_username(db): """ 同时给出用户 ID 与用户名时按「或」匹配。