refactor: finish transactional runtime migration

app/db/base.py

@@ -4,7 +4,8 @@ ORM 基类与数据访问基类。
 Base 提供声明式基类与通用的行为（字典转换、增删改查便利方法）；
 DbOper 是各业务 Oper 的基类，持有一个可注入的会话。
 """
-from typing import Any, List, Optional, Self, Union, cast
+from collections.abc import Awaitable, Callable
+from typing import Any, List, Optional, Self, TypeVar, Union, cast
 
 from sqlalchemy import (CursorResult, Executable, Identity, Integer, Sequence,
                         and_, delete, inspect, select)
@@ -13,6 +14,10 @@ from sqlalchemy.orm import DeclarativeBase, Mapped, Session, declared_attr, mapp
 
 from app.runtime.config import settings
 from app.db.decorators import async_db_query, async_db_update, db_query, db_update
+from app.db.uow import run_async_transaction, run_sync_transaction
+
+
+T = TypeVar("T")
 
 
 def execute_dml(db: Session, statement: Executable,
@@ -147,4 +152,24 @@ class DbOper:
     """
 
     def __init__(self, db: Optional[Union[Session, AsyncSession]] = None):
+        """保存调用方会话；无会话写入由组合根兼容事务执行器承接。"""
         self._db = db
+
+    def _execute_sync_write(self, operation: Callable[[Session], T]) -> T:
+        """在当前同步会话暂存，或委托组合根创建兼容事务。"""
+        if self._db is None:
+            return run_sync_transaction(operation)
+        if not isinstance(self._db, Session):
+            raise TypeError("同步写操作不能使用 AsyncSession")
+        return operation(self._db)
+
+    async def _execute_async_write(
+        self,
+        operation: Callable[[AsyncSession], Awaitable[T]],
+    ) -> T:
+        """在当前异步会话暂存，或委托组合根创建兼容事务。"""
+        if self._db is None:
+            return await run_async_transaction(operation)
+        if not isinstance(self._db, AsyncSession):
+            raise TypeError("异步写操作不能使用同步 Session")
+        return await operation(self._db)

app/db/maintenance.py

@@ -7,6 +7,7 @@ from app.db.models.downloadhistory import DownloadFiles, DownloadHistory
 from app.db.models.message import Message
 from app.db.models.siteuserdata import SiteUserData
 from app.db.models.transferhistory import TransferHistory
+from app.db.uow import SqlAlchemyUnitOfWork
 
 
 class DatabaseCleanupRepository:
@@ -20,6 +21,11 @@ class DatabaseCleanupRepository:
         """创建一次维护运行共用的数据库会话。"""
         return self._session_factory()
 
+    @staticmethod
+    def unit_of_work(db: Any) -> SqlAlchemyUnitOfWork:
+        """把当前维护 Session 适配成显式批次事务边界。"""
+        return SqlAlchemyUnitOfWork(db)
+
     @staticmethod
     def delete_messages(db: Any, cutoff: str, limit: int) -> int:
         """删除早于截止时间的消息。"""

app/db/models/agenttask.py

@@ -4,7 +4,7 @@ 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, db_update
+from app.db.decorators import db_query
 
 
 class AgentTask(Base):
@@ -49,7 +49,6 @@ class AgentTask(Base):
     )
 
     @classmethod
-    @db_update
     def add_task(cls, db: Session, **kwargs: object) -> int:
         """
         新增 Agent 定时任务并返回任务 ID。
@@ -96,7 +95,6 @@ class AgentTask(Base):
         ).scalars().all())
 
     @classmethod
-    @db_update
     def update_task(
             cls,
             db: Session,

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, db_update
+from app.db.decorators import db_query
 from app.db.models.agenttask import AgentTask
 
 
@@ -41,7 +41,6 @@ class AgentTaskRun(Base):
     )
 
     @classmethod
-    @db_update
     def begin_run(
             cls,
             db: Session,
@@ -107,7 +106,6 @@ class AgentTaskRun(Base):
         return run_id
 
     @classmethod
-    @db_update
     def finish_run(
             cls,
             db: Session,
@@ -174,7 +172,6 @@ class AgentTaskRun(Base):
         return True
 
     @classmethod
-    @db_update
     def interrupt_task(
             cls,
             db: Session,
@@ -226,7 +223,6 @@ class AgentTaskRun(Base):
         ))
 
     @classmethod
-    @db_update
     def delete_task_and_runs(
             cls,
             db: Session,

app/db/models/downloadfailure.py

@@ -4,7 +4,6 @@ from sqlalchemy import Float, Index, Integer, String, delete, select
 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_update
 from app.db.models._constraints import media_identity_constraint
 
 
@@ -115,7 +114,6 @@ class DownloadFailure(Base):
         return failure
 
     @classmethod
-    @db_update
     def delete_expired(
             cls,
             db: Session,

app/db/models/downloadhistory.py

@@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import Base, execute_dml, get_id_column
-from app.db.decorators import async_db_query, db_query, db_update
+from app.db.decorators import async_db_query, db_query
 from app.db.models._constraints import media_identity_constraint
 from app.schemas.types import MediaSource
 
@@ -295,7 +295,6 @@ class DownloadHistory(Base):
         ).scalars().all())
 
     @classmethod
-    @db_update
     def delete_before(
         cls,
         db: Session,
@@ -367,14 +366,12 @@ class DownloadFiles(Base):
         return list(db.execute(select(cls).where(cls.savepath == savepath)).scalars().all())
 
     @classmethod
-    @db_update
     def delete_by_fullpath(cls, db: Session, fullpath: str):
         db.execute(
             update(cls).where(cls.fullpath == fullpath, cls.state == 1).values(state=0)
         )
 
     @classmethod
-    @db_update
     def delete_orphans(
         cls,
         db: Session,

app/db/models/mediaserver.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import Base, execute_dml, get_id_column
-from app.db.decorators import async_db_query, db_query, db_update
+from app.db.decorators import async_db_query, db_query
 from app.db.models._constraints import media_identity_constraint
 from app.schemas.types import MediaSource
 
@@ -65,16 +65,16 @@ class MediaServerItem(Base):
         ).scalars().first()
 
     @classmethod
-    @db_update
     def empty(cls, db: Session, server: Optional[str] = None):
+        """在调用方事务中暂存媒体服务器条目清空操作。"""
         statement = delete(cls)
         if server is not None:
             statement = statement.where(cls.server == server)
         db.execute(statement, execution_options={"synchronize_session": False})
 
     @classmethod
-    @db_update
     def delete_stale(cls, db: Session, server: str, sync_time: str):
+        """在调用方事务中删除本轮同步未更新的条目。"""
         return execute_dml(
             db,
             delete(cls).where(
@@ -85,8 +85,8 @@ class MediaServerItem(Base):
         )
 
     @classmethod
-    @db_update
     def delete_excluded_servers(cls, db: Session, servers: List[str]):
+        """在调用方事务中删除不属于启用服务器的条目。"""
         statement = delete(cls)
         if servers:
             statement = statement.where(

app/db/models/message.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, execute_dml, get_id_column
-from app.db.decorators import async_db_query, db_query, db_update
+from app.db.decorators import async_db_query, db_query
 
 
 class Message(Base):
@@ -40,7 +40,6 @@ class Message(Base):
         Index('ix_message_reg_time_id', 'reg_time', 'id'),
     )
 
-    @db_update
     def create_and_to_dict(self, db: Session) -> dict:
         """
         创建消息记录并返回写入后的字段字典。
@@ -134,7 +133,6 @@ class Message(Base):
         return list(result.scalars().all())
 
     @classmethod
-    @db_update
     def delete_before(
         cls,
         db: Session,

app/db/models/passkey.py

@@ -1,11 +1,11 @@
 from typing import Optional
-from sqlalchemy import Integer, String, Boolean, DateTime, Text, select, ForeignKey
+from sqlalchemy import Integer, String, Boolean, DateTime, Text, select, ForeignKey, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 from datetime import datetime
 
 from app.db.base import Base, get_id_column
-from app.db.decorators import db_query, db_update, async_db_query, async_db_update
+from app.db.decorators import db_query, async_db_query
 
 
 class PassKey(Base):
@@ -85,19 +85,17 @@ class PassKey(Base):
         return result.scalars().first()
 
     @classmethod
-    @db_update
     def delete_by_id(cls, db: Session, passkey_id: int, user_id: int):
         """删除指定用户的PassKey"""
         passkey = db.execute(
             select(cls).where(cls.id == passkey_id, cls.user_id == user_id)
         ).scalars().first()
         if passkey:
-            passkey.delete(db, passkey.id)
+            db.delete(passkey)
             return True
         return False
 
     @classmethod
-    @async_db_update
     async def async_delete_by_id(cls, db: AsyncSession, passkey_id: int, user_id: int):
         """异步删除指定用户的PassKey"""
         result = await db.execute(
@@ -108,24 +106,22 @@ class PassKey(Base):
         )
         passkey = result.scalars().first()
         if passkey:
-            await passkey.async_delete(db, passkey.id)
+            await db.delete(passkey)
             return True
         return False
 
-    @db_update
     def update_last_used(self, db: Session, sign_count: int):
         """更新最后使用时间和签名计数"""
-        self.update(db, {
-            'last_used_at': datetime.now(),
-            'sign_count': sign_count
-        })
+        db.execute(update(type(self)).where(type(self).id == self.id).values(
+            last_used_at=datetime.now(),
+            sign_count=sign_count,
+        ))
         return True
 
-    @async_db_update
     async def async_update_last_used(self, db: AsyncSession, sign_count: int):
         """异步更新最后使用时间和签名计数"""
-        await self.async_update(db, {
-            'last_used_at': datetime.now(),
-            'sign_count': sign_count
-        })
+        await db.execute(update(type(self)).where(type(self).id == self.id).values(
+            last_used_at=datetime.now(),
+            sign_count=sign_count,
+        ))
         return True

app/db/models/plugindata.py

@@ -4,7 +4,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import get_id_column, Base
-from app.db.decorators import db_query, db_update, async_db_query
+from app.db.decorators import db_query, async_db_query
 
 
 class PluginData(Base):
@@ -49,13 +49,13 @@ class PluginData(Base):
         return result.scalar_one_or_none()
 
     @classmethod
-    @db_update
     def del_plugin_data_by_key(cls, db: Session, plugin_id: str, key: str):
+        """在调用方事务中暂存单个插件键删除。"""
         db.execute(delete(cls).where(cls.plugin_id == plugin_id, cls.key == key))
 
     @classmethod
-    @db_update
     def del_plugin_data(cls, db: Session, plugin_id: str):
+        """在调用方事务中暂存插件全部数据删除。"""
         db.execute(delete(cls).where(cls.plugin_id == plugin_id))
 
     @classmethod

app/db/models/site.py

@@ -6,7 +6,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 db_query, db_update, async_db_query, async_db_update
+from app.db.decorators import db_query, async_db_query
 
 
 class Site(Base):
@@ -102,11 +102,11 @@ class Site(Base):
         return list(db.execute(select(cls.domain).where(cls.id.in_(ids))).scalars().all())
 
     @classmethod
-    @db_update
     def reset(cls, db: Session):
+        """在调用方持有的同步事务中暂存清空操作。"""
         db.execute(delete(cls))
 
     @classmethod
-    @async_db_update
     async def async_reset(cls, db: AsyncSession):
+        """在调用方持有的异步事务中暂存清空操作。"""
         await db.execute(delete(cls))

app/db/models/sitestatistic.py

@@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import get_id_column, Base
-from app.db.decorators import db_query, db_update, async_db_query
+from app.db.decorators import db_query, async_db_query
 
 
 class SiteStatistic(Base):
@@ -41,6 +41,6 @@ class SiteStatistic(Base):
         return result.scalar_one_or_none()
 
     @classmethod
-    @db_update
     def reset(cls, db: Session):
+        """在调用方持有的事务中暂存统计表清空操作。"""
         db.execute(delete(cls))

app/db/models/siteuserdata.py

@@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import Base, execute_dml, get_id_column
-from app.db.decorators import async_db_query, db_query, db_update
+from app.db.decorators import async_db_query, db_query
 
 
 class SiteUserData(Base):
@@ -138,7 +138,6 @@ class SiteUserData(Base):
         return list(result.scalars().all())
 
     @classmethod
-    @db_update
     def delete_before(
         cls,
         db: Session,

app/db/models/systemconfig.py

@@ -4,7 +4,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 db_query, db_update, async_db_query
+from app.db.decorators import db_query, async_db_query
 
 
 class SystemConfig(Base):
@@ -28,9 +28,9 @@ class SystemConfig(Base):
         result = await db.execute(select(cls).where(cls.key == key))
         return result.scalar_one_or_none()
 
-    @db_update
     def delete_by_key(self, db: Session, key: str):
+        """在调用方持有的事务中暂存指定配置删除。"""
         systemconfig = self.get_by_key(db, key)
         if systemconfig:
-            systemconfig.delete(db, systemconfig.id)
+            db.delete(systemconfig)
         return True

app/db/models/transferhistory.py

@@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import Base, execute_dml, get_id_column
-from app.db.decorators import async_db_query, db_query, db_update
+from app.db.decorators import async_db_query, db_query
 from app.db.models._constraints import media_identity_constraint
 from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MediaSource, MediaType
 
@@ -555,14 +555,13 @@ class TransferHistory(Base):
         )).scalars().first()
 
     @classmethod
-    @db_update
     def update_download_hash(cls, db: Session, historyid: Optional[int] = None, download_hash: Optional[str] = None):
+        """在调用方事务中暂存下载任务哈希更新。"""
         db.execute(
             update(cls).where(cls.id == historyid).values(download_hash=download_hash)
         )
 
     @classmethod
-    @db_update
     def replace_by_src(cls, db: Session, **kwargs) -> "TransferHistory":
         """
         用同源存储的新记录原子替换旧整理历史。
@@ -600,7 +599,6 @@ class TransferHistory(Base):
         ).scalars().all())
 
     @classmethod
-    @db_update
     def delete_before(
         cls,
         db: Session,

app/db/models/transferpending.py

@@ -4,7 +4,7 @@ from sqlalchemy import Index, String, delete, select
 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, db_update
+from app.db.decorators import db_query
 
 
 class TransferPending(Base):
@@ -35,7 +35,6 @@ class TransferPending(Base):
     )
 
     @classmethod
-    @db_update
     def register(cls, db: Session, storage: str, src_path: str,
                  now_time: str) -> Optional["TransferPending"]:
         """
@@ -58,7 +57,6 @@ class TransferPending(Base):
         return pending
 
     @classmethod
-    @db_update
     def discard(cls, db: Session, storage: str, src_path: str) -> int:
         """
         注销一个待整理文件登记，整理到达终态（成功或失败）时调用。
@@ -93,7 +91,6 @@ class TransferPending(Base):
         ).scalars().all())
 
     @classmethod
-    @db_update
     def clear(cls, db: Session) -> int:
         """
         清空全部待整理登记。

app/db/models/user.py

@@ -4,7 +4,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 db_query, db_update, async_db_query, async_db_update
+from app.db.decorators import db_query, async_db_query
 
 
 class User(Base):
@@ -60,56 +60,46 @@ class User(Base):
         )
         return result.scalars().first()
 
-    @db_update
     def delete_by_name(self, db: Session, name: str):
         user = self.get_by_name(db, name)
         if user:
-            user.delete(db, user.id)
+            db.delete(user)
         return True
 
-    @async_db_update
     async def async_delete_by_name(self, db: AsyncSession, name: str):
         user = await self.async_get_by_name(db, name)
         if user:
-            await user.async_delete(db, user.id)
+            await db.delete(user)
         return True
 
-    @db_update
     def delete_by_id(self, db: Session, user_id: int):
         user = self.get_by_id(db, user_id)
         if user:
-            user.delete(db, user.id)
+            db.delete(user)
         return True
 
     @classmethod
-    @async_db_update
     async def async_delete_by_id(cls, db: AsyncSession, user_id: int):
         """异步按用户 ID 删除用户，供 UserOper 通过类方法调用。"""
         user = await cls.async_get_by_id(db, user_id)
         if user:
-            await user.async_delete(db, user.id)
+            await db.delete(user)
         return True
 
-    @db_update
     def update_otp_by_name(self, db: Session, name: str, otp: bool, secret: str):
         user = self.get_by_name(db, name)
         if user:
-            user.update(db, {
-                'is_otp': otp,
-                'otp_secret': secret
-            })
+            user.is_otp = otp
+            user.otp_secret = secret
             return True
         return False
 
     @classmethod
-    @async_db_update
     async def async_update_otp_by_name(cls, db: AsyncSession, name: str, otp: bool, secret: str):
         """异步按用户名更新 OTP 状态，供 UserOper 通过类方法调用。"""
         user = await cls.async_get_by_name(db, name)
         if user:
-            await user.async_update(db, {
-                'is_otp': otp,
-                'otp_secret': secret
-            })
+            user.is_otp = otp
+            user.otp_secret = secret
             return True
         return False

app/db/models/userconfig.py

@@ -3,7 +3,7 @@ from sqlalchemy import String, UniqueConstraint, JSON, select
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import get_id_column, Base
-from app.db.decorators import db_query, db_update
+from app.db.decorators import db_query
 
 
 class UserConfig(Base):
@@ -30,9 +30,9 @@ class UserConfig(Base):
             select(cls).where(cls.username == username, cls.key == key)
         ).scalars().first()
 
-    @db_update
     def delete_by_key(self, db: Session, username: str, key: str):
+        """在调用方持有的事务中暂存指定用户配置删除。"""
         userconfig = self.get_by_key(db=db, username=username, key=key)
         if userconfig:
-            userconfig.delete(db=db, rid=userconfig.id)
+            db.delete(userconfig)
         return True

app/db/models/workflow.py

@@ -7,7 +7,7 @@ from sqlalchemy.orm import Mapped, mapped_column
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db.base import Base, get_id_column
-from app.db.decorators import db_query, async_db_query, async_db_update
+from app.db.decorators import db_query, async_db_query
 
 
 class Workflow(Base):
@@ -135,8 +135,8 @@ class Workflow(Base):
         return True
 
     @classmethod
-    @async_db_update
     async def async_update_state(cls, db: AsyncSession, wid: int, state: str):
+        """在调用方持有的异步事务中暂存工作流状态。"""
         await db.execute(update(cls).where(cls.id == wid).values(state=state))
         return True
 
@@ -146,8 +146,8 @@ class Workflow(Base):
         return True
 
     @classmethod
-    @async_db_update
     async def async_start(cls, db: AsyncSession, wid: int):
+        """在调用方持有的异步事务中暂存运行中状态。"""
         await db.execute(update(cls).where(cls.id == wid).values(state='R'))
         return True
 
@@ -163,8 +163,8 @@ class Workflow(Base):
         return True
 
     @classmethod
-    @async_db_update
     async def async_fail(cls, db: AsyncSession, wid: int, result: str):
+        """在调用方持有的异步事务中暂存失败结果。"""
         await db.execute(update(cls).where(
             and_(cls.id == wid, cls.state != "P")
         ).values(
@@ -187,8 +187,8 @@ class Workflow(Base):
         return True
 
     @classmethod
-    @async_db_update
     async def async_success(cls, db: AsyncSession, wid: int, result: Optional[str] = None):
+        """在调用方持有的异步事务中暂存成功结果。"""
         await db.execute(update(cls).where(
             and_(cls.id == wid, cls.state != "P")
         ).values(
@@ -212,8 +212,8 @@ class Workflow(Base):
         return True
 
     @classmethod
-    @async_db_update
     async def async_reset(cls, db: AsyncSession, wid: int, reset_count: Optional[bool] = False):
+        """在调用方持有的异步事务中暂存执行状态重置。"""
         await db.execute(update(cls).where(cls.id == wid).values(
             state='W',
             result=None,
@@ -243,9 +243,9 @@ class Workflow(Base):
         return True
 
     @classmethod
-    @async_db_update
     async def async_update_current_action(cls, db: AsyncSession, wid: int, action_id: str, context: dict,
                                           execution_state: Optional[dict] = None):
+        """在调用方持有的异步事务中暂存动作进度。"""
         # 先获取当前current_action
         result = await db.execute(select(cls.current_action).where(cls.id == wid))
         current_action = result.scalar()

app/db/oper/agenttask.py

@@ -24,14 +24,16 @@ class AgentTaskOper(DbOper):
         新增 Agent 定时任务。
         """
         now = self._now()
-        task_id = AgentTask.add_task(
-            self._db,
-            **kwargs,
-            enabled=True,
-            last_status="waiting",
-            run_count=0,
-            created_at=now,
-            updated_at=now,
+        task_id = self._execute_sync_write(
+            lambda session: AgentTask.add_task(
+                session,
+                **kwargs,
+                enabled=True,
+                last_status="waiting",
+                run_count=0,
+                created_at=now,
+                updated_at=now,
+            )
         )
         return self.get(task_id)
 
@@ -81,38 +83,50 @@ class AgentTaskOper(DbOper):
         if not normalized_payload:
             return False
         normalized_payload["updated_at"] = self._now()
-        return AgentTask.update_task(
-            self._db,
-            task_id=task_id,
-            payload=normalized_payload,
-            user_id=user_id,
+        return self._execute_sync_write(
+            lambda session: AgentTask.update_task(
+                session,
+                task_id=task_id,
+                payload=normalized_payload,
+                user_id=user_id,
+            )
         )
 
     def delete(self, task_id: int, user_id: Optional[str] = None) -> bool:
         """
         删除非运行中的 Agent 定时任务及其运行历史。
         """
-        return AgentTaskRun.delete_task_and_runs(
-            self._db,
-            task_id=task_id,
-            user_id=user_id,
+        return self._execute_sync_write(
+            lambda session: AgentTaskRun.delete_task_and_runs(
+                session,
+                task_id=task_id,
+                user_id=user_id,
+            )
         )
 
     def begin_run(
             self,
             task_id: int,
             trigger_source: str = "scheduled",
+            *,
+            run_id: Optional[str] = None,
+            started_at: Optional[str] = None,
     ) -> Optional[AgentTaskRun]:
         """
         原子创建一次运行并返回其任务快照。
+
+        可选运行 ID 和开始时间用于恢复/幂等验证；正常调度入口由本方法生成。
         """
-        run_id = uuid4().hex
-        created_run_id = AgentTaskRun.begin_run(
-            self._db,
-            task_id=task_id,
-            run_id=run_id,
-            trigger_source=trigger_source,
-            started_at=self._now(),
+        resolved_run_id = run_id or uuid4().hex
+        resolved_started_at = started_at or self._now()
+        created_run_id = self._execute_sync_write(
+            lambda session: AgentTaskRun.begin_run(
+                session,
+                task_id=task_id,
+                run_id=resolved_run_id,
+                trigger_source=trigger_source,
+                started_at=resolved_started_at,
+            )
         )
         return self.get_run(created_run_id) if created_run_id else None
 
@@ -124,11 +138,15 @@ class AgentTaskOper(DbOper):
         """
         将遗留的运行中任务标记为中断且结果未知。
         """
-        return AgentTaskRun.interrupt_task(
-            self._db,
-            task_id=task_id,
-            result=(result or "")[:20000],
-            finished_at=self._now(),
+        finished_at = self._now()
+        normalized_result = (result or "")[:20000]
+        return self._execute_sync_write(
+            lambda session: AgentTaskRun.interrupt_task(
+                session,
+                task_id=task_id,
+                result=normalized_result,
+                finished_at=finished_at,
+            )
         )
 
     def get_run(self, run_id: str) -> Optional[AgentTaskRun]:
@@ -157,13 +175,17 @@ class AgentTaskOper(DbOper):
             disable_date_task: bool = False,
     ) -> bool:
         """收口精确运行并更新仍匹配的任务投影。"""
-        return AgentTaskRun.finish_run(
-            self._db,
-            run_id=run_id,
-            success=success,
-            result=(result or "")[:20000],
-            finished_at=self._now(),
-            disable_date_task=disable_date_task,
+        finished_at = self._now()
+        normalized_result = (result or "")[:20000]
+        return self._execute_sync_write(
+            lambda session: AgentTaskRun.finish_run(
+                session,
+                run_id=run_id,
+                success=success,
+                result=normalized_result,
+                finished_at=finished_at,
+                disable_date_task=disable_date_task,
+            )
         )
 
     def finish(

app/db/oper/downloadfailure.py

@@ -54,8 +54,10 @@ class DownloadFailureOper(DbOper):
         """
         删除已过期较久的失败记录。
         """
-        return DownloadFailure.delete_expired(
-            self._db,
-            before_time=before_time,
-            limit=limit,
+        return self._execute_sync_write(
+            lambda session: DownloadFailure.delete_expired(
+                session,
+                before_time=before_time,
+                limit=limit,
+            )
         )

app/db/oper/downloadhistory.py

@@ -127,7 +127,9 @@ class DownloadHistoryOper(DbOper):
         按fullpath删除下载文件记录
         :param fullpath: 数据key
         """
-        DownloadFiles.delete_by_fullpath(self._db, fullpath)
+        self._execute_sync_write(
+            lambda session: DownloadFiles.delete_by_fullpath(session, fullpath)
+        )
 
     def stage_delete_file_by_fullpath(self, fullpath: str) -> None:
         """暂存指定完整路径的下载文件记录删除。"""

app/db/oper/mediaserver.py

@@ -61,19 +61,32 @@ class MediaServerOper(DbOper):
         """
         清空媒体服务器数据
         """
-        MediaServerItem.empty(self._db, server)
+        self._execute_sync_write(
+            lambda session: MediaServerItem.empty(session, server)
+        )
 
     def delete_stale(self, server: str, sync_time: str) -> int:
         """
         删除本轮同步未更新的旧数据
         """
-        return MediaServerItem.delete_stale(self._db, server, sync_time)
+        return self._execute_sync_write(
+            lambda session: MediaServerItem.delete_stale(
+                session,
+                server,
+                sync_time,
+            )
+        )
 
     def delete_excluded_servers(self, servers: list[str]) -> int:
         """
         删除未启用或已移除媒体服务器的数据
         """
-        return MediaServerItem.delete_excluded_servers(self._db, servers)
+        return self._execute_sync_write(
+            lambda session: MediaServerItem.delete_excluded_servers(
+                session,
+                servers,
+            )
+        )
 
     def exists(self, **kwargs) -> Optional[MediaServerItem]:
         """

app/db/oper/message.py

@@ -62,7 +62,8 @@ class MessageOper(DbOper):
             if k not in Message.__table__.columns.keys():  # noqa
                 kwargs.pop(k)
 
-        return Message(**kwargs).create_and_to_dict(self._db)
+        message = Message(**kwargs)
+        return self._execute_sync_write(message.create_and_to_dict)
 
     async def async_add(self,
                         channel: Optional[NotificationChannel] = None,

app/db/oper/passkey.py

@@ -24,13 +24,31 @@ class PassKeyOper(DbOper):
     def create(self, payload: dict[str, Any]) -> PassKey:
         """创建 PassKey 凭证。"""
         passkey = PassKey(**payload)
-        passkey.create(self._db)
+        self._execute_sync_write(lambda session: self._stage_create(session, passkey))
         return passkey
 
+    @staticmethod
+    def _stage_create(session: Any, passkey: PassKey) -> None:
+        """在调用方事务中暂存凭证并分配主键。"""
+        session.add(passkey)
+        session.flush()
+
     def update_last_used(self, passkey: PassKey, sign_count: int) -> bool:
         """更新凭证最后使用时间和签名计数。"""
-        return bool(passkey.update_last_used(self._db, sign_count))
+        return bool(self._execute_sync_write(
+            lambda session: passkey.update_last_used(session, sign_count)
+        ))
 
     def delete_by_id(self, passkey_id: int, user_id: int) -> bool:
         """删除指定用户的凭证。"""
-        return bool(PassKey.delete_by_id(self._db, passkey_id, user_id))
+        return bool(self._execute_sync_write(
+            lambda session: PassKey.delete_by_id(session, passkey_id, user_id)
+        ))
+
+    async def async_delete_by_id(self, passkey_id: int, user_id: int) -> bool:
+        """在独立异步事务中删除指定用户的凭证。"""
+        return bool(await self._execute_async_write(
+            lambda session: PassKey.async_delete_by_id(
+                session, passkey_id, user_id
+            )
+        ))

app/db/oper/plugindata.py

@@ -80,10 +80,14 @@ class PluginDataOper(DbOper):
         :param plugin_id: 插件id
         :param key: 数据key
         """
-        if key:
-            PluginData.del_plugin_data_by_key(self._db, plugin_id, key)
-        else:
-            PluginData.del_plugin_data(self._db, plugin_id)
+        def stage(session: Session) -> None:
+            """把兼容删除入口映射到调用方或组合根持有的事务。"""
+            if key:
+                PluginData.del_plugin_data_by_key(session, plugin_id, key)
+            else:
+                PluginData.del_plugin_data(session, plugin_id)
+
+        self._execute_sync_write(stage)
 
     def stage_delete(self, plugin_id: str) -> None:
         """暂存目标插件全部数据删除并 flush，不提交调用方事务。"""

app/db/oper/site.py

@@ -116,8 +116,8 @@ class SiteOper(DbOper):
         Site.delete(self._db, sid)
 
     def reset(self) -> None:
-        """清空站点表，保留站点模型细节在数据库适配层。"""
-        Site.reset(self._db)
+        """清空站点表；兼容入口的事务由组合根统一持有。"""
+        self._execute_sync_write(Site.reset)
 
     async def stage_reset(self) -> None:
         """暂存清空站点表，由应用事务统一提交。"""

app/db/oper/transferhistory.py

@@ -264,14 +264,18 @@ class TransferHistoryOper(DbOper):
         kwargs.update({
             "date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
         })
-        TransferHistory.replace_by_src(self._db, **kwargs)
+        def stage(session: Session) -> Optional[TransferHistory]:
+            """在同一事务替换记录并返回兼容查询投影。"""
+            TransferHistory.replace_by_src(session, **kwargs)
+            return TransferHistory.get_by_src(
+                session,
+                kwargs.get("src"),
+                kwargs["src_storage"],
+            )
+
         # 保持 add_force 的既有返回契约：返回可被调用方安全读取字段的查询结果，
         # 而非事务提交后可能已脱离会话的新建实例。
-        return TransferHistory.get_by_src(
-            self._db,
-            kwargs.get("src"),
-            kwargs["src_storage"],
-        )
+        return self._execute_sync_write(stage)
 
     def stage_replace_by_src(self, **kwargs) -> TransferHistory:
         """在调用方事务内按源路径替换整理历史并返回已分配 ID 的新记录。"""
@@ -295,7 +299,13 @@ class TransferHistoryOper(DbOper):
         """
         补充转移记录download_hash
         """
-        TransferHistory.update_download_hash(self._db, historyid, download_hash)
+        self._execute_sync_write(
+            lambda session: TransferHistory.update_download_hash(
+                session,
+                historyid,
+                download_hash,
+            )
+        )
 
     def list_by_date(self, date: str) -> List[TransferHistory]:
         """

app/db/oper/transferpending.py

@@ -20,11 +20,14 @@ class TransferPendingOper(DbOper):
         :param src_path: 源文件路径
         :return: 登记记录
         """
-        return TransferPending.register(
-            self._db,
-            storage=storage,
-            src_path=src_path,
-            now_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        return self._execute_sync_write(
+            lambda session: TransferPending.register(
+                session,
+                storage=storage,
+                src_path=src_path,
+                now_time=now_time,
+            )
         )
 
     def discard(self, storage: str, src_path: str) -> int:
@@ -34,7 +37,13 @@ class TransferPendingOper(DbOper):
         :param src_path: 源文件路径
         :return: 删除的记录数
         """
-        return TransferPending.discard(self._db, storage=storage, src_path=src_path)
+        return self._execute_sync_write(
+            lambda session: TransferPending.discard(
+                session,
+                storage=storage,
+                src_path=src_path,
+            )
+        )
 
     def list_all(self, limit: Optional[int] = 5000) -> List[Tuple[str, str]]:
         """
@@ -56,4 +65,4 @@ class TransferPendingOper(DbOper):
         清空全部待整理登记。
         :return: 删除的记录数
         """
-        return TransferPending.clear(self._db)
+        return self._execute_sync_write(TransferPending.clear)

app/db/oper/user.py

@@ -11,6 +11,8 @@ runtime 兼容映射指向 SDK 薄门面；canonical 数据访问模块仍只依
 """
 from typing import List, Optional
 
+from sqlalchemy.ext.asyncio import AsyncSession
+
 from app.db.base import DbOper
 from app.db.models.user import User
 
@@ -49,27 +51,53 @@ class UserOper(DbOper):
 
     async def async_create(self, payload: dict) -> Optional[User]:
         """异步创建用户。"""
-        return await User(**payload).async_create(self._db)
+        user = User(**payload)
+
+        async def stage(session: AsyncSession) -> User:
+            """在当前异步事务中暂存用户并分配主键。"""
+            session.add(user)
+            await session.flush()
+            return user
+
+        return await self._execute_async_write(stage)
 
     async def async_update(self, user_id: int, payload: dict) -> Optional[User]:
         """异步更新用户。"""
         user = await self.async_get_by_id(user_id)
         if user:
-            await user.async_update(self._db, payload)
+            async def stage(session: AsyncSession) -> User:
+                """在当前事务中更新用户字段，必要时重新附加游离对象。"""
+                for key, value in payload.items():
+                    setattr(user, key, value)
+                return await session.merge(user)
+
+            await self._execute_async_write(stage)
         return user
 
-    async def async_delete(self, user_id: int) -> None:
+    async def async_delete(self, user_id: int) -> bool:
         """异步删除用户。"""
-        await User.async_delete_by_id(self._db, user_id)
+        return bool(await self._execute_async_write(
+            lambda session: User.async_delete_by_id(session, user_id)
+        ))
+
+    async def async_delete_by_name(self, name: str) -> bool:
+        """在独立异步事务中按用户名删除用户。"""
+        return bool(await self._execute_async_write(
+            lambda session: User().async_delete_by_name(session, name)
+        ))
 
     async def async_update_otp_by_name(
         self,
         name: str,
         otp: bool,
         secret: str,
-    ) -> None:
+    ) -> bool:
         """异步更新用户 OTP 状态。"""
-        await User.async_update_otp_by_name(self._db, name, otp, secret)
+        return bool(await self._execute_async_write(
+            lambda session: User.async_update_otp_by_name(
+                session, name, otp, secret
+            )
+        ))
 
     async def async_get_by_name(self, name: str) -> Optional[User]:
         """

app/db/uow.py

@@ -1,9 +1,65 @@
-"""SQLAlchemy 请求级事务适配器。"""
+"""SQLAlchemy 请求级事务适配器与旧 Oper 事务执行端口。"""
+
+from collections.abc import Awaitable, Callable
+from typing import Protocol, TypeVar
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
 
+T = TypeVar("T")
+
+
+class SyncTransactionRunner(Protocol):
+    """为无显式 Session 的兼容写入口提供独占同步事务。"""
+
+    def __call__(self, operation: Callable[[Session], T]) -> T:
+        """在一个独占会话中执行并提交操作。"""
+        ...
+
+
+class AsyncTransactionRunner(Protocol):
+    """为无显式 Session 的兼容写入口提供独占异步事务。"""
+
+    def __call__(
+        self,
+        operation: Callable[[AsyncSession], Awaitable[T]],
+    ) -> Awaitable[T]:
+        """在一个独占异步会话中执行并提交操作。"""
+        ...
+
+
+_sync_transaction_runner: SyncTransactionRunner | None = None
+_async_transaction_runner: AsyncTransactionRunner | None = None
+
+
+def configure_transaction_runners(
+    *,
+    sync: SyncTransactionRunner,
+    async_: AsyncTransactionRunner,
+) -> None:
+    """由组合根登记旧 Oper 兼容入口使用的显式事务执行器。"""
+    global _sync_transaction_runner, _async_transaction_runner
+    _sync_transaction_runner = sync
+    _async_transaction_runner = async_
+
+
+def run_sync_transaction(operation: Callable[[Session], T]) -> T:
+    """委托组合根在独占同步事务中执行兼容写操作。"""
+    if _sync_transaction_runner is None:
+        raise RuntimeError("同步事务执行器尚未配置")
+    return _sync_transaction_runner(operation)
+
+
+async def run_async_transaction(
+    operation: Callable[[AsyncSession], Awaitable[T]],
+) -> T:
+    """委托组合根在独占异步事务中执行兼容写操作。"""
+    if _async_transaction_runner is None:
+        raise RuntimeError("异步事务执行器尚未配置")
+    return await _async_transaction_runner(operation)
+
+
 class SqlAlchemyUnitOfWork:
     """把同步 Session 的提交与回滚能力适配为应用层事务端口。"""