diff --git a/app/api/deps.py b/app/api/deps.py index d2d922ab6..e3cc96526 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -307,7 +307,10 @@ def get_agent_chat_service( db: AsyncSession = Depends(get_async_db), ) -> AgentChatService: """组装 Agent 会话历史查询和删除服务。""" - return AgentChatService(repository=_repository("agent_chat", db)) + return AgentChatService( + repository=_repository("agent_chat", db), + unit_of_work=_transaction("async", db), + ) def get_mediaserver_query_service( diff --git a/app/application/messaging/chat.py b/app/application/messaging/chat.py index ce5664cc6..71c4545ad 100644 --- a/app/application/messaging/chat.py +++ b/app/application/messaging/chat.py @@ -45,6 +45,14 @@ class AsyncAgentChatRepository(Protocol): """删除指定服务端会话。""" ... + async def async_stage_delete( + self, + session_id: str, + user_id: Optional[str] = None, + ) -> bool: + """暂存删除指定服务端会话，不提交调用方事务。""" + ... + def get(self, session_id: str, user_id: Optional[str] = None) -> Optional[Any]: """同步读取服务端会话。""" ... @@ -83,12 +91,29 @@ class AgentChatRecord: messages: list[dict] +class AsyncUnitOfWork(Protocol): + """Agent 会话异步写用例所需的最小事务端口。""" + + async def commit(self) -> None: + """提交当前请求事务。""" + ... + + async def rollback(self) -> None: + """回滚当前请求事务。""" + ... + + class AgentChatService: """统一执行 Agent 会话查询、访问控制和删除。""" - def __init__(self, repository: AsyncAgentChatRepository) -> None: - """保存异步会话持久化端口。""" + def __init__( + self, + repository: AsyncAgentChatRepository, + unit_of_work: Optional[AsyncUnitOfWork] = None, + ) -> None: + """保存会话持久化端口和可选请求级事务。""" self._repository = repository + self._unit_of_work = unit_of_work async def list( self, @@ -137,7 +162,18 @@ class AgentChatService: record = await self.get_accessible(session_id, principal) if record is None: return False - return await self._repository.async_delete(session_id=session_id) + if self._unit_of_work is None: + return await self._repository.async_delete(session_id=session_id) + try: + deleted = await self._repository.async_stage_delete( + session_id=session_id + ) + if deleted: + await self._unit_of_work.commit() + return deleted + except Exception: + await self._unit_of_work.rollback() + raise def get_sync(self, session_id: str) -> Optional[AgentChatRecord]: """同步读取会话投影，供同步 Agent 编排路径使用。""" diff --git a/app/application/plugin/data.py b/app/application/plugin/data.py new file mode 100644 index 000000000..c7c940854 --- /dev/null +++ b/app/application/plugin/data.py @@ -0,0 +1,45 @@ +"""插件持久化数据写用例。""" + +from typing import Protocol + + +class PluginDataMutationRepository(Protocol): + """插件数据删除命令所需的无提交仓储端口。""" + + def stage_delete(self, plugin_id: str) -> None: + """暂存目标插件的全部持久化数据删除。""" + ... + + +class UnitOfWork(Protocol): + """插件数据同步写用例所需的事务端口。""" + + def commit(self) -> None: + """提交当前逻辑操作。""" + ... + + def rollback(self) -> None: + """回滚当前逻辑操作。""" + ... + + +class DeletePluginDataCommand: + """在一个显式事务中删除目标插件的全部持久化数据。""" + + def __init__( + self, + repository: PluginDataMutationRepository, + unit_of_work: UnitOfWork, + ) -> None: + """保存无提交仓储和事务所有者。""" + self._repository = repository + self._unit_of_work = unit_of_work + + def execute(self, plugin_id: str) -> None: + """暂存并提交删除；任一步失败时回滚并传播原异常。""" + try: + self._repository.stage_delete(plugin_id) + self._unit_of_work.commit() + except Exception: + self._unit_of_work.rollback() + raise diff --git a/app/db/oper/agentchat.py b/app/db/oper/agentchat.py index fdd337395..5b02427a7 100644 --- a/app/db/oper/agentchat.py +++ b/app/db/oper/agentchat.py @@ -314,6 +314,21 @@ class AgentChatOper(DbOper): await AgentChat.async_delete(self._db, chat.id) return True + async def async_stage_delete( + self, + session_id: str, + user_id: Optional[str] = None, + ) -> bool: + """暂存 Agent 会话删除并 flush，不提交请求级事务。""" + if not isinstance(self._db, AsyncSession): + raise RuntimeError("Agent 会话暂存删除需要调用方提供 AsyncSession") + chat = await self.async_get(session_id=session_id, user_id=user_id) + if not chat: + return False + await self._db.delete(chat) + await self._db.flush() + return True + @staticmethod def to_summary(chat: AgentChat) -> dict[str, Any]: """ diff --git a/app/db/oper/plugindata.py b/app/db/oper/plugindata.py index 9e3bc2184..09edfc07f 100644 --- a/app/db/oper/plugindata.py +++ b/app/db/oper/plugindata.py @@ -1,5 +1,8 @@ from typing import Any, Optional +from sqlalchemy import delete +from sqlalchemy.orm import Session + from app.db.base import DbOper from app.db.models.plugindata import PluginData @@ -82,6 +85,15 @@ class PluginDataOper(DbOper): else: PluginData.del_plugin_data(self._db, plugin_id) + def stage_delete(self, plugin_id: str) -> None: + """暂存目标插件全部数据删除并 flush，不提交调用方事务。""" + if not isinstance(self._db, Session): + raise RuntimeError("插件数据暂存删除需要调用方提供 Session") + self._db.execute( + delete(PluginData).where(PluginData.plugin_id == plugin_id) + ) + self._db.flush() + def truncate(self): """ 清空插件数据 diff --git a/app/startup/plugins_initializer.py b/app/startup/plugins_initializer.py index 6f2ce2043..73273774c 100644 --- a/app/startup/plugins_initializer.py +++ b/app/startup/plugins_initializer.py @@ -18,6 +18,7 @@ from app.runtime.extensions.plugin_manager import ( ) from app.runtime.extensions.plugin.dependency import PluginDependencyInstallResult from app.application.plugin.catalog import PluginCatalogService +from app.application.plugin.data import DeletePluginDataCommand from app.adapters.external.plugin.client import PluginMarketClient from app.runtime.extensions.plugin.storage import ( PluginStorage, @@ -41,6 +42,8 @@ from app.adapters.system.plugin.package import PluginPackageManager from app.adapters.system.host import SystemUtils from app.db.oper.plugindata import PluginDataOper from app.db.oper.systemconfig import SystemConfigOper +from app.db.session import SessionFactory +from app.db.uow import SqlAlchemyUnitOfWork from app.runtime.log import logger from app.foundation.version import compare_version from app.schemas.plugin import PluginRuntimeStatus @@ -52,6 +55,18 @@ async def _async_write_plugin_config(key, value): return await SystemConfigOper().async_set(key, value) +def _delete_plugin_data(plugin_id: str) -> None: + """用独占同步会话执行插件重置的数据删除事务。""" + session = SessionFactory() + try: + DeletePluginDataCommand( + repository=PluginDataOper(session), + unit_of_work=SqlAlchemyUnitOfWork(session), + ).execute(plugin_id) + finally: + session.close() + + def _prepare_legacy_plugin_import(*, plugin_id: str, plugin_dir: Path) -> None: """在执行旧插件顶层代码前准备其静态导入所需的宿主资源。""" for capability_id in scan_plugin_resource_imports(plugin_id, plugin_dir): @@ -98,7 +113,7 @@ def configure_plugin_services() -> None: write=lambda key, value: SystemConfigOper().set(key, value), async_write=_async_write_plugin_config, delete=lambda key: SystemConfigOper().delete(key), - delete_data=lambda plugin_id: PluginDataOper().del_data(plugin_id), + delete_data=_delete_plugin_data, )) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 801119102..9fd051335 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -367,6 +367,9 @@ flowchart LR `application/subscription/write.py` 决定事务与 post-commit 边界，`SubscribeOper.stage_add()` 只查重、`add` 和 `flush`。旧 SDK 显式构造的无会话 Oper 暂留兼容自动短会话，不得被新代码复用。 `transaction-debt-baseline.json` 将存量 178 个 Model 事务装饰器冻结为只降不增低水位。 +- 站点、历史、工作流、Agent 会话删除和插件数据重置已经形成同构事务切片；对应 Application + Command/Service 持有 UoW，Oper 的 `stage_*` 方法只修改当前会话。插件数据重置从 + `startup/plugins_initializer.py` 创建独占会话，插件直接使用 `PluginDataOper` 的旧 ABI 仅作兼容。 - 每次表结构变更必须新增 `database/versions/` 下的 Alembic 迁移。 - 运行期业务配置使用 `SystemConfigKey` 枚举 + `SystemConfigOper`，禁止裸字符串键； 用户级配置使用 `UserConfigOper`。 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 66498573e..35574e020 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0（ARCH-201～203）、阶段 1（ARCH-210～212）与 ARCH-220～221 已完成，后续任务按 ID 独立提交和回滚 +> 实施进度：阶段 0（ARCH-201～203）、阶段 1（ARCH-210～212）与阶段 2（ARCH-220～222）已完成，后续任务按 ID 独立提交和回滚 ## 1. 结论先行 @@ -399,6 +399,20 @@ flowchart TB 每个切片沿用 ARCH-221，不允许批量移动全部 Model 方法。查询方法可在写边界稳定后再迁移。 +**实施记录（2026-08-21）**： + +| 风险域 | 规范事务入口 | 结果 | +| --- | --- | --- | +| 站点配置 | `SiteMutationCommand` + Async UoW | create/update/priorities/delete/reset 均先 stage 再 commit | +| 下载/整理历史 | `DownloadHistoryMutationCommand`、`TransferHistoryMutationCommand` + Sync UoW | 多表删除与文件副作用顺序已有聚焦回归 | +| 工作流 | `WorkflowMutationCommand`、`WorkflowDefinitionCommand` + Sync/Async UoW | 定义写入提交后才刷新 timer/event | +| Agent chat | `AgentChatService` + 请求级 Async UoW | API 会话删除改为 `async_stage_delete()`；失败回滚、缺失不提交 | +| 插件数据重置 | `DeletePluginDataCommand` + 独占 Sync Session/UoW | `PluginDataOper.stage_delete()` 只 DELETE/flush；重置链由 startup 装配 | + +旧插件与宿主存量代码直接构造 `PluginDataOper`、`AgentChatOper` 的行为继续保留；新 API 和插件 +重置链不得回退到这些自动提交兼容方法。五类矩阵聚焦测试共 57 项通过，事务 ratchet 仍为 +178 且没有新增或搬移 Model 装饰器。 + ### 阶段 3：类型化运行时装配，减少全局服务定位 #### ARCH-230：建立类型化 HostRuntime / AppState diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index 0d71fe03b..674063524 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -104,6 +104,10 @@ Oper classes accept and return persistence values. Turning a `MediaInfo` or UoW and post-commit callback, and `SubscribeOper.stage_add()` only queries, adds, and flushes. Preserve `SubscribeOper.add()` only for legacy SDK callers; new host code must not use that auto-commit compatibility path. +- The same rule applies to `SiteMutationCommand`, history/workflow commands, + `AgentChatService.delete()`, and `DeletePluginDataCommand`: bind the repository + and UoW to one request/operation Session. Legacy plugin-facing Oper methods may + remain temporarily, but a new endpoint or startup workflow must call `stage_*`. Run `./.venv/bin/python scripts/architecture/baseline.py --check-host` after persistence changes. A deliberate debt reduction may refresh the low-water mark diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 6d7660350..b1df6c7e7 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6092, - "edge_sha256": "1935d43d8d3c0c3e3687109f0b81a4e56cbc4bc2ed4763a706e2b0d7c119f378", + "edge_count": 6095, + "edge_sha256": "06c09c175ac007c7ef891e2a25f5036c8bc3993a817837c03824bb591168ea73", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -5890,12 +5890,15 @@ "app.startup.plugins_initializer -> app.application", "app.startup.plugins_initializer -> app.application.plugin", "app.startup.plugins_initializer -> app.application.plugin.catalog", + "app.startup.plugins_initializer -> app.application.plugin.data", "app.startup.plugins_initializer -> app.application.plugin.routes", "app.startup.plugins_initializer -> app.application.site", "app.startup.plugins_initializer -> app.db", "app.startup.plugins_initializer -> app.db.oper", "app.startup.plugins_initializer -> app.db.oper.plugindata", "app.startup.plugins_initializer -> app.db.oper.systemconfig", + "app.startup.plugins_initializer -> app.db.session", + "app.startup.plugins_initializer -> app.db.uow", "app.startup.plugins_initializer -> app.foundation", "app.startup.plugins_initializer -> app.foundation.version", "app.startup.plugins_initializer -> app.runtime", @@ -6109,7 +6112,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 757, + "module_count": 758, "modules": [ "app", "app.adapters", @@ -6379,6 +6382,7 @@ "app.application.plugin", "app.application.plugin.catalog", "app.application.plugin.config", + "app.application.plugin.data", "app.application.plugin.folders", "app.application.plugin.install", "app.application.plugin.routes", diff --git a/tests/test_agent_chat_transaction.py b/tests/test_agent_chat_transaction.py new file mode 100644 index 000000000..aa3ee1705 --- /dev/null +++ b/tests/test_agent_chat_transaction.py @@ -0,0 +1,92 @@ +"""Agent 会话删除的请求级事务边界测试。""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.application.messaging.chat import AgentChatService + + +def _principal() -> SimpleNamespace: + """构造拥有目标会话的普通用户。""" + return SimpleNamespace(id=1, name="alice", is_superuser=False) + + +def _chat() -> SimpleNamespace: + """构造应用服务投影所需的最小会话记录。""" + return SimpleNamespace( + id=9, + session_id="session-9", + client_session_id=None, + title="事务会话", + channel=None, + source=None, + user_id="1", + username="alice", + original_chat_id=None, + message_count=0, + created_at=None, + updated_at=None, + display_messages=[], + ) + + +@pytest.mark.asyncio +async def test_delete_stages_then_commits_once() -> None: + """有权限的会话删除只能由 Service 在暂存成功后提交一次。""" + calls: list[str] = [] + repository = Mock() + repository.async_get = AsyncMock(return_value=_chat()) + repository.async_stage_delete = AsyncMock( + side_effect=lambda **_kwargs: calls.append("stage") or True + ) + unit_of_work = Mock() + unit_of_work.commit = AsyncMock( + side_effect=lambda: calls.append("commit") + ) + unit_of_work.rollback = AsyncMock() + service = AgentChatService(repository, unit_of_work) + + assert await service.delete("session-9", _principal()) is True + + assert calls == ["stage", "commit"] + unit_of_work.rollback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delete_rolls_back_flush_failure() -> None: + """暂存删除失败必须回滚并传播原异常。""" + error = RuntimeError("flush failed") + repository = Mock() + repository.async_get = AsyncMock(return_value=_chat()) + repository.async_stage_delete = AsyncMock(side_effect=error) + unit_of_work = Mock() + unit_of_work.commit = AsyncMock() + unit_of_work.rollback = AsyncMock() + service = AgentChatService(repository, unit_of_work) + + with pytest.raises(RuntimeError) as raised: + await service.delete("session-9", _principal()) + + assert raised.value is error + unit_of_work.commit.assert_not_awaited() + unit_of_work.rollback.assert_awaited_once_with() + + +@pytest.mark.asyncio +async def test_delete_missing_chat_does_not_open_write_transaction() -> None: + """会话不存在时保持旧 False 返回，且不执行 stage 或 commit。""" + repository = Mock() + repository.async_get = AsyncMock(return_value=None) + repository.async_stage_delete = AsyncMock() + unit_of_work = Mock() + unit_of_work.commit = AsyncMock() + unit_of_work.rollback = AsyncMock() + service = AgentChatService(repository, unit_of_work) + + assert await service.delete("missing", _principal()) is False + + repository.async_stage_delete.assert_not_awaited() + unit_of_work.commit.assert_not_awaited() + unit_of_work.rollback.assert_not_awaited() diff --git a/tests/test_plugin_data_transaction.py b/tests/test_plugin_data_transaction.py new file mode 100644 index 000000000..a2bb6531f --- /dev/null +++ b/tests/test_plugin_data_transaction.py @@ -0,0 +1,60 @@ +"""插件持久化数据删除的事务所有权测试。""" + +from unittest.mock import Mock + +import pytest + +from app.application.plugin.data import DeletePluginDataCommand +from app.db.models.plugindata import PluginData +from app.db.oper.plugindata import PluginDataOper + + +def test_delete_plugin_data_stages_then_commits() -> None: + """插件重置必须在仓储暂存完成后由 Application Command 提交。""" + calls: list[str] = [] + repository = Mock() + repository.stage_delete.side_effect = lambda _plugin_id: calls.append("stage") + unit_of_work = Mock() + unit_of_work.commit.side_effect = lambda: calls.append("commit") + command = DeletePluginDataCommand(repository, unit_of_work) + + command.execute("Demo") + + assert calls == ["stage", "commit"] + repository.stage_delete.assert_called_once_with("Demo") + unit_of_work.rollback.assert_not_called() + + +def test_delete_plugin_data_rolls_back_commit_failure() -> None: + """插件数据删除提交失败必须回滚并保留原异常。""" + error = RuntimeError("commit failed") + repository = Mock() + unit_of_work = Mock() + unit_of_work.commit.side_effect = error + command = DeletePluginDataCommand(repository, unit_of_work) + + with pytest.raises(RuntimeError) as raised: + command.execute("Demo") + + assert raised.value is error + unit_of_work.rollback.assert_called_once_with() + + +def test_plugin_data_oper_stage_delete_does_not_commit(db, monkeypatch) -> None: + """Oper 只暂存目标插件删除，其他插件数据与提交权均不受影响。""" + db.add( + PluginData(plugin_id="Target", key="one", value=1), + PluginData(plugin_id="Other", key="two", value=2), + ) + commit = Mock(wraps=db.session.commit) + monkeypatch.setattr(db.session, "commit", commit) + oper = PluginDataOper(db.session) + + oper.stage_delete("Target") + + remaining = PluginData.get_plugin_data(db.session, "Other") + deleted = PluginData.get_plugin_data(db.session, "Target") + assert [item.key for item in remaining] == ["two"] + assert deleted == [] + commit.assert_not_called() + db.session.rollback()