refactor: isolate agent history queries

This commit is contained in:
jxxghp
2026-08-23 14:21:53 +08:00
parent d8b8583d49
commit 8fc9a021b0
11 changed files with 121 additions and 61 deletions
+5 -5
View File
@@ -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,
+3 -3
View File
@@ -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,
+14 -8
View File
@@ -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(
+10 -6
View File
@@ -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(
+1 -1
View File
@@ -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 的提交权不会被底层抢走。
@@ -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` 个异步)。站点、消息、用户、订阅、下载/整理历史、工作流、MediaServerSiteUserData 的宿主 Oper 已迁到显式 Session 路径;对应旧插件 Model 调用由独立 `legacy_*` 外壳保留,可同时接受显式 Session 与无 Session 的位置/关键字参数。剩余正式装饰器仍会隐式创建会话,查询返回的 ORM 对象也可能跨层流转,后续继续按 AgentChat、AgentTaskRun、TransferPending、SystemConfig、PassKey 和 SubscribeHistory 等风险切片迁移。
3. **查询侧数据库兼容 ABI 仍未完全收口。** 写事务装饰器已降为 `0`,正式 `db_query/async_db_query` 已降至 `12` 个(`5` 个同步、`7` 个异步)。站点、消息、用户、订阅、下载/整理历史、工作流、MediaServerSiteUserData、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 |
+1 -1
View File
@@ -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
+3 -33
View File
@@ -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",
+30
View File
@@ -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")
+2 -2
View File
@@ -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": []}
@@ -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 与用户名时按「或」匹配。