refactor: close transactional boundary debt batch

app/db/adapters/chain.py

@@ -19,7 +19,12 @@ from app.application.chain.events import (
     snapshot_transfer_result,
     transfer_result_event_key,
 )
-from app.application.history import TransferHistoryRecord, TransferHistoryWriter
+from app.application.history import (
+    DownloadFileWrite,
+    DownloadHistoryWrite,
+    TransferHistoryRecord,
+    TransferHistoryWriter,
+)
 from app.application.outbox import (
     DOWNLOAD_ADDED_TOPIC,
     DurableEventCommand,
@@ -32,7 +37,10 @@ from app.application.transfer.execution import (
     TransferExecutionState,
     TransferSettlementResult,
 )
-from app.db.adapters.outbox import SqlAlchemyOutboxRepository
+from app.db.adapters.outbox import (
+    SqlAlchemyOutboxDispatchStore,
+    SqlAlchemyOutboxStager,
+)
 from app.db.models.transfersettlementreceipt import TransferSettlementReceipt
 from app.db.oper.downloadhistory import DownloadHistoryOper
 from app.db.oper.transferexecutionstep import TransferExecutionStepOper
@@ -114,8 +122,8 @@ class TransactionalChainDurableEventWriter(ChainDurableEventWriter):
     def download_added(
         self,
         *,
-        history_payload: dict[str, Any],
-        file_payloads: list[dict[str, Any]],
+        history: DownloadHistoryWrite,
+        files: tuple[DownloadFileWrite, ...],
         event_payload: dict[str, Any],
         after_commit: Callable[[], None],
         publish: Callable[[dict[str, Any]], None],
@@ -124,17 +132,20 @@ class TransactionalChainDurableEventWriter(ChainDurableEventWriter):
         session = self._session_factory()
         try:
             repository = DownloadHistoryOper(session)
-            outbox = SqlAlchemyOutboxRepository(session)
+            outbox = SqlAlchemyOutboxStager(session)
             command = DurableEventCommand(
                 unit_of_work=SqlAlchemyUnitOfWork(session),
-                outbox=outbox,
+                stager=outbox,
+                store=SqlAlchemyOutboxDispatchStore(self._session_factory),
             )
             def stage_business() -> int:
                 """在同一事务暂存下载历史和可选文件清单。"""
-                history = repository.stage_add(history_payload)
-                if file_payloads:
-                    repository.stage_add_files(file_payloads)
-                return int(history.id)
+                record = repository.stage_add(history.to_payload())
+                if files:
+                    repository.stage_add_files([
+                        file_item.to_payload() for file_item in files
+                    ])
+                return int(record.id)
 
             def build_intent(history_id: int) -> OutboxIntent:
                 """历史 ID 确定后构造本次下载事实的稳定事件键。"""
@@ -182,7 +193,8 @@ class TransactionalChainDurableEventWriter(ChainDurableEventWriter):
 
             command = DurableEventCommand(
                 unit_of_work=SqlAlchemyUnitOfWork(session),
-                outbox=SqlAlchemyOutboxRepository(session),
+                stager=SqlAlchemyOutboxStager(session),
+                store=SqlAlchemyOutboxDispatchStore(self._session_factory),
             )
 
             def stage_business() -> _StagedTransferResult:
@@ -271,7 +283,7 @@ class TransactionalChainDurableEventWriter(ChainDurableEventWriter):
                 )
 
             try:
-                result = command.execute(
+                execution = command.execute(
                     intent=build_intent if topic is not None else None,
                     stage_business=stage_business,
                     publish=(
@@ -301,6 +313,7 @@ class TransactionalChainDurableEventWriter(ChainDurableEventWriter):
                 if replay is None:
                     raise
                 return replay
+            result = execution.value
             return result.settlement or result.history
         finally:
             session.close()

app/db/adapters/configuration.py

@@ -0,0 +1,91 @@
+"""用户配置快照的显式短会话与事务适配器。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Optional, Union
+
+from sqlalchemy.orm import Session
+
+from app.db.oper.userconfig import UserConfigOper
+from app.db.uow import SqlAlchemyUnitOfWork
+from app.schemas.common import JsonData
+from app.schemas.types import UserConfigKey
+
+
+class TransactionalUserConfigurationRepository:
+    """在短事务提交后发布用户配置缓存，并在发布失败时重载事实源。"""
+
+    def __init__(
+        self,
+        session_factory: Callable[[], Session],
+        snapshot: Optional[UserConfigOper] = None,
+    ) -> None:
+        """保存会话工厂及进程级用户配置快照。"""
+        self._session_factory = session_factory
+        self._snapshot = snapshot or UserConfigOper()
+
+    def load_snapshot(self) -> None:
+        """使用独立只读会话从数据库发布完整配置快照。"""
+        with self._session_factory() as session:
+            self._snapshot.load_snapshot(session)
+
+    def get(
+        self,
+        username: str,
+        key: Union[str, UserConfigKey],
+    ) -> JsonData:
+        """从进程级快照读取一项深拷贝配置。"""
+        return self._snapshot.get(username=username, key=key)
+
+    def set(
+        self,
+        username: str,
+        key: Union[str, UserConfigKey],
+        value: JsonData,
+    ) -> None:
+        """提交一项配置后发布快照，发布异常时从数据库恢复快照。"""
+        with self._snapshot.write_scope():
+            with self._session_factory() as session:
+                unit_of_work = SqlAlchemyUnitOfWork(session)
+                try:
+                    deleted = self._snapshot.stage_set(
+                        session,
+                        username,
+                        key,
+                        value,
+                    )
+                    unit_of_work.commit()
+                except Exception:
+                    unit_of_work.rollback()
+                    raise
+            try:
+                self._snapshot.publish(
+                    username,
+                    key,
+                    value,
+                    deleted=deleted,
+                )
+            except Exception:
+                self.load_snapshot()
+                raise
+
+    def publish_rename(self, previous_name: str, current_name: str) -> None:
+        """发布已提交用户改名，并在同一写锁内以数据库事实源收口。"""
+        with self._snapshot.write_scope():
+            try:
+                self._snapshot.publish_rename(previous_name, current_name)
+            except Exception:
+                self.load_snapshot()
+                raise
+            self.load_snapshot()
+
+    def publish_delete(self, username: str) -> None:
+        """发布已提交用户删除，并在同一写锁内以数据库事实源收口。"""
+        with self._snapshot.write_scope():
+            try:
+                self._snapshot.publish_delete(username)
+            except Exception:
+                self.load_snapshot()
+                raise
+            self.load_snapshot()

app/db/adapters/history/__init__.py

@@ -0,0 +1 @@
+"""历史持久化适配器包。"""

app/db/adapters/history/download.py

@@ -0,0 +1,287 @@
+"""下载历史的类型化查询、写入与事务适配器。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from contextlib import AbstractAsyncContextManager
+from copy import deepcopy
+from typing import Optional, TypeVar, Union
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
+
+from app.application.history import (
+    DownloadFileSnapshot,
+    DownloadFileWrite,
+    DownloadHistorySnapshot,
+    DownloadHistoryWrite,
+)
+from app.db.oper.downloadhistory import DownloadHistoryOper
+from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
+from app.schemas.media import normalize_media_source
+from app.schemas.types import MediaSource
+
+ResultT = TypeVar("ResultT")
+
+
+def _project_history(record: object) -> DownloadHistorySnapshot:
+    """在 Session 内把下载历史 ORM 记录投影为不可变快照。"""
+    history_id = getattr(record, "id", None)
+    path = getattr(record, "path", None)
+    media_type = getattr(record, "type", None)
+    title = getattr(record, "title", None)
+    if (
+        not isinstance(history_id, int)
+        or not isinstance(path, str)
+        or not isinstance(media_type, str)
+        or not isinstance(title, str)
+    ):
+        raise ValueError("下载历史记录缺少稳定身份、路径、类型或标题")
+    media_source = normalize_media_source(getattr(record, "media_source", None))
+    media_id_value = getattr(record, "media_id", None)
+    media_id = str(media_id_value).strip() if media_id_value is not None else None
+    if not media_source or not media_id or media_id == "0":
+        media_source = None
+        media_id = None
+    return DownloadHistorySnapshot(
+        id=history_id,
+        path=path,
+        type=media_type,
+        title=title,
+        year=getattr(record, "year", None),
+        media_source=media_source,
+        media_id=media_id,
+        music_type=getattr(record, "music_type", None),
+        seasons=getattr(record, "seasons", None),
+        episodes=getattr(record, "episodes", None),
+        image=getattr(record, "image", None),
+        poster=getattr(record, "poster", None),
+        downloader=getattr(record, "downloader", None),
+        download_hash=getattr(record, "download_hash", None),
+        torrent_name=getattr(record, "torrent_name", None),
+        torrent_description=getattr(record, "torrent_description", None),
+        torrent_site=getattr(record, "torrent_site", None),
+        userid=(
+            str(userid_value)
+            if (userid_value := getattr(record, "userid", None)) is not None
+            else None
+        ),
+        username=getattr(record, "username", None),
+        channel=getattr(record, "channel", None),
+        date=getattr(record, "date", None),
+        note=deepcopy(getattr(record, "note", None)),
+        media_category=getattr(record, "media_category", None),
+        episode_group=getattr(record, "episode_group", None),
+        custom_words=getattr(record, "custom_words", None),
+    )
+
+
+def _project_file(record: object) -> DownloadFileSnapshot:
+    """在 Session 内把下载文件 ORM 记录投影为不可变快照。"""
+    file_id = getattr(record, "id", None)
+    state = getattr(record, "state", None)
+    if not isinstance(file_id, int) or not isinstance(state, int):
+        raise ValueError("下载文件记录缺少稳定身份或状态")
+    return DownloadFileSnapshot(
+        id=file_id,
+        downloader=getattr(record, "downloader", None),
+        download_hash=getattr(record, "download_hash", None),
+        fullpath=getattr(record, "fullpath", None),
+        savepath=getattr(record, "savepath", None),
+        filepath=getattr(record, "filepath", None),
+        torrentname=getattr(record, "torrentname", None),
+        state=state,
+    )
+
+
+class TransactionalDownloadHistoryRepository:
+    """为 Chain 和 Agent 下载历史读写创建短生命周期 Session。"""
+
+    def __init__(
+        self,
+        *,
+        sync_session: Callable[[], Session],
+        async_session: Callable[[], AbstractAsyncContextManager[AsyncSession]],
+    ) -> None:
+        """保存由启动组合根提供的同步与异步 Session 工厂。"""
+        self._sync_session = sync_session
+        self._async_session = async_session
+
+    def _read(self, operation: Callable[[DownloadHistoryOper], ResultT]) -> ResultT:
+        """在独立同步 Session 中执行一次只读操作。"""
+        session = self._sync_session()
+        try:
+            return operation(DownloadHistoryOper(session))
+        finally:
+            session.close()
+
+    def get_by_hash(
+        self,
+        download_hash: str,
+    ) -> Optional[DownloadHistorySnapshot]:
+        """按下载任务 Hash 返回最新历史快照。"""
+        return self._read(
+            lambda repository: (
+                _project_history(record)
+                if (record := repository.get_by_hash(download_hash)) is not None
+                else None
+            )
+        )
+
+    def get_by_hashes(
+        self,
+        download_hashes: list[str],
+    ) -> dict[str, DownloadHistorySnapshot]:
+        """批量返回以下载任务 Hash 为键的最新历史快照。"""
+        return self._read(
+            lambda repository: {
+                download_hash: _project_history(record)
+                for download_hash, record in repository.get_by_hashes(download_hashes).items()
+            }
+        )
+
+    def get_by_path(self, path: str) -> Optional[DownloadHistorySnapshot]:
+        """按下载保存路径返回历史快照。"""
+        return self._read(
+            lambda repository: (
+                _project_history(record)
+                if (record := repository.get_by_path(path)) is not None
+                else None
+            )
+        )
+
+    def get_by_media_identity(
+        self,
+        media_source: MediaSource,
+        media_id: str,
+        music_type: Optional[str] = None,
+    ) -> list[DownloadHistorySnapshot]:
+        """按规范媒体身份返回历史快照。"""
+        return self._read(
+            lambda repository: [
+                _project_history(record)
+                for record in repository.get_by_media_identity(
+                    media_source=media_source,
+                    media_id=media_id,
+                    music_type=music_type,
+                )
+            ]
+        )
+
+    def get_file_by_fullpath(
+        self,
+        fullpath: str,
+    ) -> Optional[DownloadFileSnapshot]:
+        """按完整路径返回一条有效下载文件快照。"""
+        return self._read(
+            lambda repository: (
+                _project_file(record)
+                if (record := repository.get_file_by_fullpath(fullpath)) is not None
+                else None
+            )
+        )
+
+    def get_files_by_hash(
+        self,
+        download_hash: str,
+        state: Optional[int] = None,
+    ) -> list[DownloadFileSnapshot]:
+        """按下载任务 Hash 返回文件快照。"""
+        return self._read(
+            lambda repository: [
+                _project_file(record)
+                for record in repository.get_files_by_hash(
+                    download_hash,
+                    state=state,
+                )
+            ]
+        )
+
+    def get_files_by_savepath(self, savepath: str) -> list[DownloadFileSnapshot]:
+        """按保存目录返回下载文件快照。"""
+        return self._read(
+            lambda repository: [
+                _project_file(record)
+                for record in repository.get_files_by_savepath(savepath)
+            ]
+        )
+
+    async def async_list_by_page(
+        self,
+        page: int = 1,
+        count: int = 30,
+    ) -> list[DownloadHistorySnapshot]:
+        """在独立异步 Session 内分页读取并投影历史。"""
+        async with self._async_session() as session:
+            records = await DownloadHistoryOper(session).async_list_by_page(
+                page,
+                count,
+            )
+            return [_project_history(record) for record in records]
+
+    def add(
+        self,
+        history: DownloadHistoryWrite,
+        files: tuple[DownloadFileWrite, ...] = (),
+    ) -> int:
+        """在一个同步事务中新增历史与关联文件。"""
+        session = self._sync_session()
+        unit_of_work = SqlAlchemyUnitOfWork(session)
+        try:
+            repository = DownloadHistoryOper(session)
+            record = repository.stage_add(history.to_payload())
+            if files:
+                repository.stage_add_files([file_item.to_payload() for file_item in files])
+            history_id = int(record.id)
+            unit_of_work.commit()
+            return history_id
+        except Exception:
+            unit_of_work.rollback()
+            raise
+        finally:
+            session.close()
+
+    async def async_delete(self, history_id: int) -> None:
+        """在一个异步事务中删除指定下载历史。"""
+        async with self._async_session() as session:
+            unit_of_work = SqlAlchemyAsyncUnitOfWork(session)
+            try:
+                await DownloadHistoryOper(session).async_delete_history(history_id)
+                await unit_of_work.commit()
+            except Exception:
+                await unit_of_work.rollback()
+                raise
+
+
+class SessionDownloadHistoryRepository:
+    """把 API 请求持有的 Session 适配为下载历史查询和暂存端口。"""
+
+    def __init__(self, session: Union[Session, AsyncSession]) -> None:
+        """保存由请求依赖独占的数据库 Session。"""
+        self._session = session
+
+    async def async_list_by_page(
+        self,
+        page: int = 1,
+        count: int = 30,
+    ) -> list[DownloadHistorySnapshot]:
+        """在请求异步 Session 内分页读取并投影历史。"""
+        if not isinstance(self._session, AsyncSession):
+            raise RuntimeError("下载历史异步查询需要 AsyncSession")
+        records = await DownloadHistoryOper(self._session).async_list_by_page(
+            page,
+            count,
+        )
+        return [_project_history(record) for record in records]
+
+    def stage_delete_history(self, history_id: int) -> None:
+        """在请求同步 Session 内暂存下载历史删除。"""
+        if not isinstance(self._session, Session):
+            raise RuntimeError("下载历史同步删除需要 Session")
+        DownloadHistoryOper(self._session).stage_delete_history(history_id)
+
+    def stage_delete_file_by_fullpath(self, fullpath: str) -> None:
+        """在请求同步 Session 内暂存下载文件失效状态。"""
+        if not isinstance(self._session, Session):
+            raise RuntimeError("下载文件同步变更需要 Session")
+        DownloadHistoryOper(self._session).stage_delete_file_by_fullpath(fullpath)

app/db/adapters/outbox.py

@@ -1,10 +1,15 @@
-"""Application outbox 端口的 SQLAlchemy 持久化适配器。"""
+"""Application outbox 暂存与派发端口的 SQLAlchemy 适配器。"""
 
+from collections.abc import Callable
+from contextlib import AbstractAsyncContextManager
 from datetime import datetime
+from typing import Optional
 
 from sqlalchemy import or_, select, update
-from sqlalchemy.orm import Session
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
+from sqlalchemy.sql import Select
+from sqlalchemy.sql.dml import Update
 
 from app.application.outbox import ClaimedOutboxMessage, OutboxIntent
 from app.db.base import execute_dml
@@ -16,198 +21,278 @@ def _iso(value: datetime) -> str:
     return value.isoformat()
 
 
-class SqlAlchemyOutboxRepository:
-    """使用调用方 Session 原子暂存并条件认领 outbox。"""
+def _message(model: OutboxMessage, attempt: int) -> ClaimedOutboxMessage:
+    """把已认领 ORM 行复制为脱离会话的稳定消息。"""
+    return ClaimedOutboxMessage(
+        message_id=model.id,
+        event_key=model.event_key,
+        topic=model.topic,
+        payload=dict(model.payload),
+        payload_version=model.payload_version,
+        attempt=attempt,
+    )
+
+
+def _claim_query(
+    now_text: str,
+    event_key: Optional[str] = None,
+) -> Select[tuple[OutboxMessage]]:
+    """构造到期且 lease 可取得的候选查询。"""
+    statement = select(OutboxMessage).where(
+        OutboxMessage.status.in_(("pending", "processing")),
+        OutboxMessage.next_retry_at <= now_text,
+        or_(
+            OutboxMessage.lease_until.is_(None),
+            OutboxMessage.lease_until <= now_text,
+        ),
+    )
+    if event_key is not None:
+        statement = statement.where(OutboxMessage.event_key == event_key)
+    return statement.order_by(OutboxMessage.id).limit(1)
+
+
+def _claim_update(
+    candidate: OutboxMessage,
+    now_text: str,
+    lease_until: datetime,
+) -> Update:
+    """构造带旧 attempt fencing 的条件认领更新。"""
+    return (
+        update(OutboxMessage)
+        .where(
+            OutboxMessage.id == candidate.id,
+            OutboxMessage.attempt == candidate.attempt,
+            OutboxMessage.status.in_(("pending", "processing")),
+            OutboxMessage.next_retry_at <= now_text,
+            or_(
+                OutboxMessage.lease_until.is_(None),
+                OutboxMessage.lease_until <= now_text,
+            ),
+        )
+        .values(
+            status="processing",
+            attempt=candidate.attempt + 1,
+            lease_until=_iso(lease_until),
+        )
+    )
+
+
+class SqlAlchemyOutboxStager:
+    """只在调用方同步业务事务中暂存 durable intent。"""
 
     def __init__(self, session: Session) -> None:
-        """保存由调用方拥有的 SQLAlchemy Session。"""
+        """保存业务事务拥有的同步 Session。"""
         self._session = session
 
     def stage(self, intent: OutboxIntent, now: datetime) -> None:
-        """加入当前事务并 flush，使唯一键冲突在业务 commit 前暴露。"""
-        self._session.add(
-            OutboxMessage(
-                event_key=intent.event_key,
-                topic=intent.topic,
-                payload_version=intent.payload_version,
-                payload=intent.payload,
-                status="pending",
-                attempt=0,
-                next_retry_at=_iso(now),
-                created_at=_iso(now),
-            )
-        )
+        """加入当前事务并 flush，使唯一键冲突在 commit 前暴露。"""
+        self._session.add(_outbox_model(intent, now))
         self._session.flush()
 
+
+class SqlAlchemyAsyncOutboxStager:
+    """只在调用方异步业务事务中暂存 durable intent。"""
+
+    def __init__(self, session: AsyncSession) -> None:
+        """保存业务事务拥有的异步 Session。"""
+        self._session = session
+
+    async def stage(self, intent: OutboxIntent, now: datetime) -> None:
+        """暂存并 flush，业务行与 intent 由同一次 commit 决定。"""
+        self._session.add(_outbox_model(intent, now))
+        await self._session.flush()
+
+
+class SqlAlchemyOutboxDispatchStore:
+    """用独立同步短事务认领并结算 outbox 消息。"""
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """保存每次操作创建独立 Session 的工厂。"""
+        self._session_factory = session_factory
+
     def claim(
         self,
         now: datetime,
         lease_until: datetime,
-    ) -> ClaimedOutboxMessage | None:
-        """条件更新候选行；并发丢失竞争时返回 None。"""
-        now_text = _iso(now)
-        candidate = self._session.execute(
-            select(OutboxMessage)
-            .where(
-                OutboxMessage.status.in_(("pending", "processing")),
-                OutboxMessage.next_retry_at <= now_text,
-                or_(
-                    OutboxMessage.lease_until.is_(None),
-                    OutboxMessage.lease_until <= now_text,
-                ),
-            )
-            .order_by(OutboxMessage.id)
-            .limit(1)
-        ).scalars().first()
-        if candidate is None:
-            return None
-        next_attempt = candidate.attempt + 1
-        claimed = execute_dml(
-            self._session,
-            update(OutboxMessage)
-            .where(
-                OutboxMessage.id == candidate.id,
-                OutboxMessage.attempt == candidate.attempt,
-                or_(
-                    OutboxMessage.lease_until.is_(None),
-                    OutboxMessage.lease_until <= now_text,
-                ),
-            )
-            .values(
-                status="processing",
-                attempt=next_attempt,
-                lease_until=_iso(lease_until),
-            ),
-        )
-        self._session.commit()
-        if not claimed:
-            return None
-        return ClaimedOutboxMessage(
-            message_id=candidate.id,
-            event_key=candidate.event_key,
-            topic=candidate.topic,
-            payload=dict(candidate.payload),
-            payload_version=candidate.payload_version,
-            attempt=next_attempt,
-        )
+    ) -> Optional[ClaimedOutboxMessage]:
+        """原子认领最早一条到期消息。"""
+        return self._claim(now, lease_until)
 
     def claim_by_event_key(
         self,
         event_key: str,
         now: datetime,
         lease_until: datetime,
-    ) -> bool:
-        """按事件键原子认领同步投递，避免与 dispatcher 并发重复发送。"""
+    ) -> Optional[ClaimedOutboxMessage]:
+        """按稳定事件键原子认领到期消息。"""
+        return self._claim(now, lease_until, event_key)
+
+    def _claim(
+        self,
+        now: datetime,
+        lease_until: datetime,
+        event_key: Optional[str] = None,
+    ) -> Optional[ClaimedOutboxMessage]:
+        """在独立事务中以 compare-and-swap 取得 lease。"""
         now_text = _iso(now)
-        candidate = self._session.execute(
-            select(OutboxMessage)
-            .where(
-                OutboxMessage.event_key == event_key,
-                OutboxMessage.status.in_(("pending", "processing")),
-                OutboxMessage.next_retry_at <= now_text,
-                or_(
-                    OutboxMessage.lease_until.is_(None),
-                    OutboxMessage.lease_until <= now_text,
+        with self._session_factory() as session:
+            candidate = session.execute(_claim_query(now_text, event_key)).scalars().first()
+            if candidate is None:
+                return None
+            next_attempt = candidate.attempt + 1
+            claimed = execute_dml(
+                session,
+                _claim_update(candidate, now_text, lease_until),
+            )
+            session.commit()
+            return _message(candidate, next_attempt) if claimed else None
+
+    def complete(
+        self,
+        message_id: int,
+        attempt: int,
+        completed_at: datetime,
+    ) -> bool:
+        """仅允许当前 attempt 的 processing owner 标记完成。"""
+        with self._session_factory() as session:
+            changed = execute_dml(
+                session,
+                update(OutboxMessage)
+                .where(
+                    OutboxMessage.id == message_id,
+                    OutboxMessage.status == "processing",
+                    OutboxMessage.attempt == attempt,
+                )
+                .values(
+                    status="completed",
+                    completed_at=_iso(completed_at),
+                    lease_until=None,
                 ),
             )
-            .limit(1)
-        ).scalars().first()
-        if candidate is None:
-            return False
-        claimed = execute_dml(
-            self._session,
-            update(OutboxMessage)
-            .where(
-                OutboxMessage.id == candidate.id,
-                OutboxMessage.attempt == candidate.attempt,
-                OutboxMessage.event_key == event_key,
-                OutboxMessage.status.in_(("pending", "processing")),
-                OutboxMessage.next_retry_at <= now_text,
-                or_(
-                    OutboxMessage.lease_until.is_(None),
-                    OutboxMessage.lease_until <= now_text,
-                ),
-            )
-            .values(
-                status="processing",
-                attempt=OutboxMessage.attempt + 1,
-                lease_until=_iso(lease_until),
-            ),
-        )
-        self._session.commit()
-        return bool(claimed)
-
-    def complete(self, message_id: int, completed_at: datetime) -> None:
-        """持久化完成终态并释放 lease。"""
-        self._session.execute(
-            update(OutboxMessage)
-            .where(OutboxMessage.id == message_id)
-            .values(status="completed", completed_at=_iso(completed_at), lease_until=None)
-        )
-        self._session.commit()
-
-    def complete_by_event_key(self, event_key: str, completed_at: datetime) -> None:
-        """即时 post-commit 全部成功时按幂等键收口对应 intent。"""
-        self._session.execute(
-            update(OutboxMessage)
-            .where(OutboxMessage.event_key == event_key)
-            .values(status="completed", completed_at=_iso(completed_at), lease_until=None)
-        )
-        self._session.commit()
+            session.commit()
+            return bool(changed)
 
     def retry(
         self,
         message_id: int,
+        attempt: int,
         *,
         next_retry_at: datetime,
         last_error: str,
         dead: bool,
+    ) -> bool:
+        """仅允许当前 attempt 的 owner 释放 lease 或写入 dead 终态。"""
+        with self._session_factory() as session:
+            changed = execute_dml(
+                session,
+                update(OutboxMessage)
+                .where(
+                    OutboxMessage.id == message_id,
+                    OutboxMessage.status == "processing",
+                    OutboxMessage.attempt == attempt,
+                )
+                .values(
+                    status="dead" if dead else "pending",
+                    next_retry_at=_iso(next_retry_at),
+                    lease_until=None,
+                    last_error=last_error,
+                ),
+            )
+            session.commit()
+            return bool(changed)
+
+
+class SqlAlchemyAsyncOutboxDispatchStore:
+    """用独立异步短事务认领并结算 outbox 消息。"""
+
+    def __init__(
+        self,
+        session_factory: Callable[[], AbstractAsyncContextManager[AsyncSession]],
     ) -> None:
-        """持久化下一次退避或不可自动重试的 dead 终态。"""
-        self._session.execute(
-            update(OutboxMessage)
-            .where(OutboxMessage.id == message_id)
-            .values(
-                status="dead" if dead else "pending",
-                next_retry_at=_iso(next_retry_at),
-                lease_until=None,
-                last_error=last_error,
-            )
-        )
-        self._session.commit()
+        """保存每次操作创建独立异步 Session 的工厂。"""
+        self._session_factory = session_factory
 
-class SqlAlchemyAsyncOutboxStager:
-    """只负责把 outbox 意图加入调用方异步事务。"""
-
-    def __init__(self, session: AsyncSession) -> None:
-        """保存由异步订阅命令拥有的 Session。"""
-        self._session = session
-
-    async def stage(self, intent: OutboxIntent, now: datetime) -> None:
-        """暂存并 flush，确保业务行与意图由同一次 commit 决定。"""
-        self._session.add(
-            OutboxMessage(
-                event_key=intent.event_key,
-                topic=intent.topic,
-                payload_version=intent.payload_version,
-                payload=intent.payload,
-                status="pending",
-                attempt=0,
-                next_retry_at=_iso(now),
-                created_at=_iso(now),
-            )
-        )
-        await self._session.flush()
-
-    async def complete_by_event_key(
+    async def claim_by_event_key(
         self,
         event_key: str,
+        now: datetime,
+        lease_until: datetime,
+    ) -> Optional[ClaimedOutboxMessage]:
+        """按稳定事件键原子认领到期消息。"""
+        now_text = _iso(now)
+        async with self._session_factory() as session:
+            candidate = (await session.execute(_claim_query(now_text, event_key))).scalars().first()
+            if candidate is None:
+                return None
+            next_attempt = candidate.attempt + 1
+            result = await session.execute(_claim_update(candidate, now_text, lease_until))
+            await session.commit()
+            return _message(candidate, next_attempt) if result.rowcount else None
+
+    async def complete(
+        self,
+        message_id: int,
+        attempt: int,
         completed_at: datetime,
-    ) -> None:
-        """异步 post-commit 全部成功时按幂等键收口 intent。"""
-        await self._session.execute(
-            update(OutboxMessage)
-            .where(OutboxMessage.event_key == event_key)
-            .values(status="completed", completed_at=_iso(completed_at), lease_until=None)
-        )
-        await self._session.commit()
+    ) -> bool:
+        """仅允许当前 attempt 的 processing owner 标记完成。"""
+        async with self._session_factory() as session:
+            result = await session.execute(
+                update(OutboxMessage)
+                .where(
+                    OutboxMessage.id == message_id,
+                    OutboxMessage.status == "processing",
+                    OutboxMessage.attempt == attempt,
+                )
+                .values(
+                    status="completed",
+                    completed_at=_iso(completed_at),
+                    lease_until=None,
+                )
+            )
+            await session.commit()
+            return bool(result.rowcount)
+
+    async def retry(
+        self,
+        message_id: int,
+        attempt: int,
+        *,
+        next_retry_at: datetime,
+        last_error: str,
+        dead: bool,
+    ) -> bool:
+        """仅允许当前 attempt 的 owner 释放 lease 或写入 dead 终态。"""
+        async with self._session_factory() as session:
+            result = await session.execute(
+                update(OutboxMessage)
+                .where(
+                    OutboxMessage.id == message_id,
+                    OutboxMessage.status == "processing",
+                    OutboxMessage.attempt == attempt,
+                )
+                .values(
+                    status="dead" if dead else "pending",
+                    next_retry_at=_iso(next_retry_at),
+                    lease_until=None,
+                    last_error=last_error,
+                )
+            )
+            await session.commit()
+            return bool(result.rowcount)
+
+
+def _outbox_model(intent: OutboxIntent, now: datetime) -> OutboxMessage:
+    """构造由业务事务持有的新 outbox ORM 行。"""
+    payload = dict(intent.payload)
+    payload["idempotency_key"] = intent.event_key
+    return OutboxMessage(
+        event_key=intent.event_key,
+        topic=intent.topic,
+        payload_version=intent.payload_version,
+        payload=payload,
+        status="pending",
+        attempt=0,
+        next_retry_at=_iso(now),
+        created_at=_iso(now),
+    )

app/db/adapters/subscription.py

@@ -2,12 +2,19 @@
 
 from collections.abc import Callable
 from contextlib import AbstractAsyncContextManager
-from datetime import datetime, timezone
-from typing import Any
+from datetime import datetime, timedelta, timezone
+from typing import Any, Optional
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
+from app.application.outbox import (
+    OUTBOX_LEASE_SECONDS,
+    AsyncOutboxDispatchStore,
+    ClaimedOutboxMessage,
+    OutboxDispatchStore,
+    OutboxLeaseLostError,
+)
 from app.application.subscription.write import (
     AfterCommitEffect,
     AsyncAfterCommitEffect,
@@ -18,8 +25,10 @@ from app.application.subscription.write import (
     subscription_added_report_key,
 )
 from app.db.adapters.outbox import (
+    SqlAlchemyAsyncOutboxDispatchStore,
     SqlAlchemyAsyncOutboxStager,
-    SqlAlchemyOutboxRepository,
+    SqlAlchemyOutboxDispatchStore,
+    SqlAlchemyOutboxStager,
 )
 from app.db.oper.subscribe import SubscribeOper
 from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
@@ -51,7 +60,8 @@ class TransactionalSubscribeWriter:
         """在独占同步会话内执行一次完整订阅新增事务。"""
         session = self._sync_session()
         try:
-            outbox = SqlAlchemyOutboxRepository(session)
+            outbox = SqlAlchemyOutboxStager(session)
+            dispatch_store = SqlAlchemyOutboxDispatchStore(self._sync_session)
             command = CreateSubscriptionCommand(
                 repository=SubscribeOper(session),
                 unit_of_work=SqlAlchemyUnitOfWork(session),
@@ -61,21 +71,13 @@ class TransactionalSubscribeWriter:
             def delivered(subscribe_id: int) -> None:
                 """执行提交后编排，分别收口已确认的 durable intent。"""
                 if after_commit:
-                    report_delivered = after_commit(subscribe_id)
-                    outbox.complete_by_event_key(
-                        subscription_added_event_key(subscribe_id, payload),
-                        datetime.now(timezone.utc),
+                    _deliver_added_effects(
+                        dispatch_store,
+                        subscribe_id,
+                        payload,
+                        notification,
+                        lambda: after_commit(subscribe_id),
                     )
-                    if notification:
-                        outbox.complete_by_event_key(
-                            subscription_added_notification_key(subscribe_id, payload),
-                            datetime.now(timezone.utc),
-                        )
-                    if report_delivered is not False:
-                        outbox.complete_by_event_key(
-                            subscription_added_report_key(subscribe_id, payload),
-                            datetime.now(timezone.utc),
-                        )
 
             return command.execute(
                 identity,
@@ -98,6 +100,7 @@ class TransactionalSubscribeWriter:
         """在独占异步会话作用域内执行一次完整订阅新增事务。"""
         async with self._async_session() as session:
             outbox = SqlAlchemyAsyncOutboxStager(session)
+            dispatch_store = SqlAlchemyAsyncOutboxDispatchStore(self._async_session)
             command = AsyncCreateSubscriptionCommand(
                 repository=SubscribeOper(session),
                 unit_of_work=SqlAlchemyAsyncUnitOfWork(session),
@@ -107,21 +110,13 @@ class TransactionalSubscribeWriter:
             async def delivered(subscribe_id: int) -> None:
                 """异步执行提交后编排，分别收口已确认的 durable intent。"""
                 if after_commit:
-                    report_delivered = await after_commit(subscribe_id)
-                    await outbox.complete_by_event_key(
-                        subscription_added_event_key(subscribe_id, payload),
-                        datetime.now(timezone.utc),
+                    await _deliver_added_effects_async(
+                        dispatch_store,
+                        subscribe_id,
+                        payload,
+                        notification,
+                        lambda: after_commit(subscribe_id),
                     )
-                    if notification:
-                        await outbox.complete_by_event_key(
-                            subscription_added_notification_key(subscribe_id, payload),
-                            datetime.now(timezone.utc),
-                        )
-                    if report_delivered is not False:
-                        await outbox.complete_by_event_key(
-                            subscription_added_report_key(subscribe_id, payload),
-                            datetime.now(timezone.utc),
-                        )
 
             return await command.execute(
                 identity,
@@ -130,3 +125,160 @@ class TransactionalSubscribeWriter:
                 delivered,
                 notification,
             )
+
+
+def _added_effect_keys(
+    subscribe_id: int,
+    payload: dict[str, Any],
+    notification: Optional[dict[str, object]],
+) -> tuple[str, ...]:
+    """返回组合回调实际包含的独立 durable effect 键。"""
+    keys = [subscription_added_event_key(subscribe_id, payload)]
+    if notification:
+        keys.append(subscription_added_notification_key(subscribe_id, payload))
+    keys.append(subscription_added_report_key(subscribe_id, payload))
+    return tuple(keys)
+
+
+def _claim_added_effects(
+    store: OutboxDispatchStore,
+    keys: tuple[str, ...],
+    now: datetime,
+) -> Optional[tuple[ClaimedOutboxMessage, ...]]:
+    """全量认领组合回调；竞争丢失时释放本次已取得的 lease。"""
+    claimed: list[ClaimedOutboxMessage] = []
+    for key in keys:
+        message = store.claim_by_event_key(
+            key,
+            now,
+            now + timedelta(seconds=OUTBOX_LEASE_SECONDS),
+        )
+        if message is None:
+            for owned in claimed:
+                store.retry(
+                    owned.message_id,
+                    owned.attempt,
+                    next_retry_at=now,
+                    last_error="组合副作用由其他 owner 接管",
+                    dead=False,
+                )
+            return None
+        claimed.append(message)
+    return tuple(claimed)
+
+
+def _deliver_added_effects(
+    store: OutboxDispatchStore,
+    subscribe_id: int,
+    payload: dict[str, Any],
+    notification: Optional[dict[str, object]],
+    effect: Callable[[], Optional[bool]],
+) -> None:
+    """认领组合回调并按事件、通知、统计的确认结果分别结算。"""
+    now = datetime.now(timezone.utc)
+    claimed = _claim_added_effects(
+        store,
+        _added_effect_keys(subscribe_id, payload, notification),
+        now,
+    )
+    if claimed is None:
+        return
+    try:
+        report_delivered = effect()
+    except Exception as error:
+        for message in claimed:
+            store.retry(
+                message.message_id,
+                message.attempt,
+                next_retry_at=now,
+                last_error=str(error)[:4000],
+                dead=False,
+            )
+        raise
+    for message in claimed[:-1]:
+        if not store.complete(message.message_id, message.attempt, now):
+            raise OutboxLeaseLostError("订阅新增完成凭证已失效")
+    report = claimed[-1]
+    if report_delivered is False:
+        store.retry(
+            report.message_id,
+            report.attempt,
+            next_retry_at=now,
+            last_error="订阅新增统计未确认",
+            dead=False,
+        )
+    else:
+        if not store.complete(report.message_id, report.attempt, now):
+            raise OutboxLeaseLostError("订阅新增统计完成凭证已失效")
+
+
+async def _claim_added_effects_async(
+    store: AsyncOutboxDispatchStore,
+    keys: tuple[str, ...],
+    now: datetime,
+) -> Optional[tuple[ClaimedOutboxMessage, ...]]:
+    """异步全量认领组合回调，竞争丢失时释放已取得 lease。"""
+    claimed: list[ClaimedOutboxMessage] = []
+    for key in keys:
+        message = await store.claim_by_event_key(
+            key,
+            now,
+            now + timedelta(seconds=OUTBOX_LEASE_SECONDS),
+        )
+        if message is None:
+            for owned in claimed:
+                await store.retry(
+                    owned.message_id,
+                    owned.attempt,
+                    next_retry_at=now,
+                    last_error="组合副作用由其他 owner 接管",
+                    dead=False,
+                )
+            return None
+        claimed.append(message)
+    return tuple(claimed)
+
+
+async def _deliver_added_effects_async(
+    store: AsyncOutboxDispatchStore,
+    subscribe_id: int,
+    payload: dict[str, Any],
+    notification: Optional[dict[str, object]],
+    effect: Callable[[], Any],
+) -> None:
+    """异步认领组合回调并按各 intent 的确认结果分别结算。"""
+    now = datetime.now(timezone.utc)
+    claimed = await _claim_added_effects_async(
+        store,
+        _added_effect_keys(subscribe_id, payload, notification),
+        now,
+    )
+    if claimed is None:
+        return
+    try:
+        report_delivered = await effect()
+    except Exception as error:
+        for message in claimed:
+            await store.retry(
+                message.message_id,
+                message.attempt,
+                next_retry_at=now,
+                last_error=str(error)[:4000],
+                dead=False,
+            )
+        raise
+    for message in claimed[:-1]:
+        if not await store.complete(message.message_id, message.attempt, now):
+            raise OutboxLeaseLostError("订阅新增完成凭证已失效")
+    report = claimed[-1]
+    if report_delivered is False:
+        await store.retry(
+            report.message_id,
+            report.attempt,
+            next_retry_at=now,
+            last_error="订阅新增统计未确认",
+            dead=False,
+        )
+    else:
+        if not await store.complete(report.message_id, report.attempt, now):
+            raise OutboxLeaseLostError("订阅新增统计完成凭证已失效")

app/db/adapters/user.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 
 from collections.abc import Callable, Mapping
 from contextlib import AbstractAsyncContextManager
-from typing import Any
+from typing import Any, Optional, cast
 
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
@@ -13,9 +15,12 @@ from app.application.security.user import (
     AuxiliaryUserCreate,
     ChainUserRepository,
     FrozenJson,
+    LastActiveSuperuserError,
     UserAuthSnapshot,
+    UserNameConflictError,
     UserRepository,
     UserSnapshot,
+    UserUpdateResult,
 )
 from app.db.models.user import User
 from app.db.oper.user import UserOper
@@ -54,12 +59,12 @@ class SqlAlchemyUserRepository(UserRepository):
         self._session = session
         self._oper = UserOper(db=session)
 
-    def get_by_name(self, name: str) -> UserSnapshot | None:
+    def get_by_name(self, name: str) -> Optional[UserSnapshot]:
         """在同步请求会话中按用户名读取冻结快照。"""
         model = self._oper.get_by_name(name)
         return _to_snapshot(model) if model else None
 
-    def get_by_id(self, user_id: int) -> UserSnapshot | None:
+    def get_by_id(self, user_id: int) -> Optional[UserSnapshot]:
         """在同步请求会话中按 ID 读取冻结快照。"""
         model = self._oper.get_by_id(user_id)
         return _to_snapshot(model) if model else None
@@ -68,33 +73,64 @@ class SqlAlchemyUserRepository(UserRepository):
         """在异步请求会话中读取全部冻结用户快照。"""
         return [_to_snapshot(model) for model in await self._oper.async_list()]
 
-    async def async_get_by_name(self, name: str) -> UserSnapshot | None:
+    async def async_get_by_name(self, name: str) -> Optional[UserSnapshot]:
         """在异步请求会话中按用户名读取冻结快照。"""
         model = await self._oper.async_get_by_name(name)
         return _to_snapshot(model) if model else None
 
-    async def async_get_by_id(self, user_id: int) -> UserSnapshot | None:
+    async def async_get_by_id(self, user_id: int) -> Optional[UserSnapshot]:
         """在异步请求会话中按 ID 读取冻结快照。"""
         model = await self._oper.async_get_by_id(user_id)
         return _to_snapshot(model) if model else None
 
-    async def async_create(self, payload: dict[str, Any]) -> UserSnapshot | None:
+    async def async_create(
+        self,
+        payload: dict[str, Any],
+    ) -> Optional[UserSnapshot]:
         """在请求事务中暂存用户创建并返回冻结快照。"""
-        model = await self._oper.async_create(payload)
-        return _to_snapshot(model) if model else None
+        session = self._require_async_session()
+        model = User(**payload)
+        session.add(model)
+        try:
+            await session.flush()
+        except IntegrityError as error:
+            raise UserNameConflictError(payload.get("name")) from error
+        return _to_snapshot(model)
 
     async def async_update(
         self,
         user_id: int,
         payload: dict[str, Any],
-    ) -> UserSnapshot | None:
-        """在请求事务中暂存用户更新并返回更新后的冻结快照。"""
-        model = await self._oper.async_update(user_id, payload)
-        return _to_snapshot(model) if model else None
+    ) -> Optional[UserUpdateResult]:
+        """原子更新用户；数据库外键负责按用户名级联偏好。"""
+        session = self._require_async_session()
+        model = await self._locked_user(session, user_id, payload)
+        if model is None:
+            return None
+        old_name = model.name
+        new_name = str(payload.get("name", old_name))
+        values = {key: value for key, value in payload.items() if key != "id"}
+        for key, value in values.items():
+            setattr(model, key, value)
+        try:
+            await session.flush()
+        except IntegrityError as error:
+            raise UserNameConflictError(new_name) from error
+        return UserUpdateResult(
+            user=_to_snapshot(model),
+            previous_name=old_name,
+        )
 
-    async def async_delete(self, user_id: int) -> None:
-        """在请求事务中暂存用户删除。"""
-        await self._oper.async_delete(user_id)
+    async def async_delete(self, user_id: int) -> Optional[str]:
+        """原子删除用户；数据库外键负责级联偏好和 PassKey。"""
+        session = self._require_async_session()
+        model = await self._locked_user(session, user_id, None)
+        if model is None:
+            return None
+        username: str = model.name
+        await session.delete(model)
+        await session.flush()
+        return username
 
     async def async_update_otp_by_name(
         self,
@@ -105,6 +141,45 @@ class SqlAlchemyUserRepository(UserRepository):
         """在请求事务中暂存用户 OTP 状态更新。"""
         await self._oper.async_update_otp_by_name(name, otp, secret)
 
+    def _require_async_session(self) -> AsyncSession:
+        """返回写用例要求的异步 Session，拒绝错误组合。"""
+        if not isinstance(self._session, AsyncSession):
+            raise RuntimeError("用户异步写入必须绑定 AsyncSession")
+        return self._session
+
+    @staticmethod
+    async def _locked_user(
+        session: AsyncSession,
+        user_id: int,
+        payload: Optional[dict[str, Any]],
+    ) -> Optional[User]:
+        """先锁管理员集合再锁目标用户，保护并发下最后一个启用管理员。"""
+        result = await session.execute(
+            select(User)
+            .where(User.is_active.is_(True), User.is_superuser.is_(True))
+            .order_by(User.id)
+            .with_for_update()
+        )
+        administrators = list(result.scalars().all())
+        model = next(
+            (administrator for administrator in administrators if administrator.id == user_id),
+            None,
+        )
+        if model is None:
+            locked = await session.execute(select(User).where(User.id == user_id).with_for_update())
+            model = cast(Optional[User], locked.scalars().first())
+            if model is None:
+                return None
+        remains_active = (
+            payload is not None
+            and bool(payload.get("is_active", model.is_active))
+            and bool(payload.get("is_superuser", model.is_superuser))
+        )
+        removes_active_superuser = bool(model.is_active and model.is_superuser and not remains_active)
+        if removes_active_superuser and len(administrators) <= 1:
+            raise LastActiveSuperuserError(model.name)
+        return model
+
 
 class TransactionalUserRepository(ChainUserRepository):
     """为 Chain、Agent 和进程级认证提供短生命周期用户会话。"""
@@ -119,23 +194,23 @@ class TransactionalUserRepository(ChainUserRepository):
         self._sync_session = sync_session
         self._async_session = async_session
 
-    def get_by_name(self, name: str) -> UserSnapshot | None:
+    def get_by_name(self, name: str) -> Optional[UserSnapshot]:
         """按用户名读取公开用户快照。"""
         with self._sync_session() as session:
             return SqlAlchemyUserRepository(session).get_by_name(name)
 
-    def get_by_id(self, user_id: int) -> UserSnapshot | None:
+    def get_by_id(self, user_id: int) -> Optional[UserSnapshot]:
         """按 ID 读取公开用户快照。"""
         with self._sync_session() as session:
             return SqlAlchemyUserRepository(session).get_by_id(user_id)
 
-    def get_auth_by_name(self, name: str) -> UserAuthSnapshot | None:
+    def get_auth_by_name(self, name: str) -> Optional[UserAuthSnapshot]:
         """按用户名读取认证凭据快照。"""
         with self._sync_session() as session:
             model = UserOper(db=session).get_by_name(name)
             return _to_auth_snapshot(model) if model else None
 
-    async def async_get_by_name(self, name: str) -> UserSnapshot | None:
+    async def async_get_by_name(self, name: str) -> Optional[UserSnapshot]:
         """异步按用户名读取公开用户快照。"""
         async with self._async_session() as session:
             return await SqlAlchemyUserRepository(session).async_get_by_name(name)
@@ -164,7 +239,7 @@ class TransactionalUserRepository(ChainUserRepository):
     def get_notification_settings(
         self,
         name: str,
-    ) -> Mapping[str, FrozenJson] | None:
+    ) -> Optional[Mapping[str, FrozenJson]]:
         """同步读取用户通知设置的只读快照。"""
         user = self.get_by_name(name)
         return user.settings if user else None
@@ -172,12 +247,15 @@ class TransactionalUserRepository(ChainUserRepository):
     async def async_get_notification_settings(
         self,
         name: str,
-    ) -> Mapping[str, FrozenJson] | None:
+    ) -> Optional[Mapping[str, FrozenJson]]:
         """异步读取用户通知设置的只读快照。"""
         user = await self.async_get_by_name(name)
         return user.settings if user else None
 
-    def find_name_by_bindings(self, bindings: Mapping[str, object]) -> str | None:
+    def find_name_by_bindings(
+        self,
+        bindings: Mapping[str, object],
+    ) -> Optional[str]:
         """仅在全部绑定唯一匹配同一启用用户时返回用户名。"""
         if not bindings:
             return None