refactor: migrate high-risk write transactions

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(

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 编排路径使用。"""

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

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]:
         """

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):
         """
         清空插件数据

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,
     ))
 
 

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`。

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

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

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",

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()

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()