refactor: own subscription create transactions

app/application/subscription/write.py

@@ -15,7 +15,7 @@ app/application/history.py 里整理历史的写入路径同构。
 下方 _translate 单点承担，两条链路只在「怎么查、怎么写」上分叉。
 """
 
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from typing import Optional, Protocol, Tuple
 
 from app.domain.context import MediaInfo, MusicInfo
@@ -26,6 +26,9 @@ from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType
 # 而后续按身份去重也会失效，所以必须在查询与建模之前短路
 INCOMPLETE_IDENTITY = (0, "媒体身份不完整")
 
+AfterCommitEffect = Callable[[int], None]
+AsyncAfterCommitEffect = Callable[[int], Awaitable[None]]
+
 
 class SubscribeWriter(Protocol):
     """订阅写入应用服务使用的数据端口。"""
@@ -35,16 +38,151 @@ class SubscribeWriter(Protocol):
         identity: dict,
         payload: dict,
         username: Optional[str] = None,
+        after_commit: Optional[AfterCommitEffect] = None,
     ) -> Tuple[int, str]:
-        """同步新增订阅。"""
+        """同步新增订阅，并在事务成功后执行外部副作用。"""
 
     async def async_add(
         self,
         identity: dict,
         payload: dict,
         username: Optional[str] = None,
+        after_commit: Optional[AsyncAfterCommitEffect] = None,
     ) -> Tuple[int, str]:
-        """异步新增订阅。"""
+        """异步新增订阅，并在事务成功后执行外部副作用。"""
+
+
+class StagedSubscription(Protocol):
+    """订阅仓储暂存结果的结构化端口，避免 Application 反向约束适配器类型。"""
+
+    @property
+    def subscribe_id(self) -> int:
+        """返回已创建或已存在的订阅 ID。"""
+        ...
+
+    @property
+    def message(self) -> str:
+        """返回兼容旧入口的结果说明。"""
+        ...
+
+    @property
+    def created(self) -> bool:
+        """标识本次是否暂存了一条新记录。"""
+        ...
+
+
+class SubscriptionStagingRepository(Protocol):
+    """新增订阅命令需要的无提交仓储端口。"""
+
+    def stage_add(
+        self,
+        identity: dict,
+        payload: dict,
+        username: Optional[str] = None,
+    ) -> StagedSubscription:
+        """暂存同步新增，命中重复订阅时不写入。"""
+        ...
+
+    async def async_stage_add(
+        self,
+        identity: dict,
+        payload: dict,
+        username: Optional[str] = None,
+    ) -> StagedSubscription:
+        """暂存异步新增，命中重复订阅时不写入。"""
+        ...
+
+
+class UnitOfWork(Protocol):
+    """同步订阅新增命令使用的最小事务端口。"""
+
+    def commit(self) -> None:
+        """提交当前逻辑操作。"""
+        ...
+
+    def rollback(self) -> None:
+        """回滚当前逻辑操作。"""
+        ...
+
+
+class AsyncUnitOfWork(Protocol):
+    """异步订阅新增命令使用的最小事务端口。"""
+
+    async def commit(self) -> None:
+        """提交当前逻辑操作。"""
+        ...
+
+    async def rollback(self) -> None:
+        """回滚当前逻辑操作。"""
+        ...
+
+
+class CreateSubscriptionCommand:
+    """暂存并提交一条同步订阅，重复请求保持历史返回且不产生提交。"""
+
+    def __init__(
+        self,
+        repository: SubscriptionStagingRepository,
+        unit_of_work: UnitOfWork,
+    ) -> None:
+        """注入无提交仓储和事务所有者。"""
+        self._repository = repository
+        self._unit_of_work = unit_of_work
+
+    def execute(
+        self,
+        identity: dict,
+        payload: dict,
+        username: Optional[str] = None,
+        after_commit: Optional[AfterCommitEffect] = None,
+    ) -> Tuple[int, str]:
+        """执行同步新增；事务失败回滚，提交后副作用失败不反向回滚。"""
+        try:
+            staged = self._repository.stage_add(identity, payload, username)
+            if staged.created:
+                self._unit_of_work.commit()
+        except Exception:
+            self._unit_of_work.rollback()
+            raise
+        if staged.subscribe_id and after_commit:
+            after_commit(staged.subscribe_id)
+        return staged.subscribe_id, staged.message
+
+
+class AsyncCreateSubscriptionCommand:
+    """暂存并提交一条异步订阅，事务成功后才把结果交给副作用调用方。"""
+
+    def __init__(
+        self,
+        repository: SubscriptionStagingRepository,
+        unit_of_work: AsyncUnitOfWork,
+    ) -> None:
+        """注入无提交异步仓储和事务所有者。"""
+        self._repository = repository
+        self._unit_of_work = unit_of_work
+
+    async def execute(
+        self,
+        identity: dict,
+        payload: dict,
+        username: Optional[str] = None,
+        after_commit: Optional[AsyncAfterCommitEffect] = None,
+    ) -> Tuple[int, str]:
+        """执行异步新增；事务失败回滚，提交后副作用失败不反向回滚。"""
+        try:
+            staged = await self._repository.async_stage_add(
+                identity,
+                payload,
+                username,
+            )
+            if staged.created:
+                await self._unit_of_work.commit()
+        except Exception:
+            await self._unit_of_work.rollback()
+            raise
+        if staged.subscribe_id and after_commit:
+            await after_commit(staged.subscribe_id)
+        return staged.subscribe_id, staged.message
 
 
 _configured_subscribe_writer: Callable[[], SubscribeWriter] | None = None
@@ -130,6 +268,7 @@ def _translate(
 def add_subscribe(
     mediainfo: MediaInfo | MusicInfo,
     subscribe_oper: Optional[SubscribeWriter] = None,
+    after_commit: Optional[AfterCommitEffect] = None,
     **kwargs,
 ) -> Tuple[int, str]:
     """
@@ -137,6 +276,7 @@ def add_subscribe(
 
     :param mediainfo: 识别结果
     :param subscribe_oper: 复用的订阅操作对象，未传时由启动组合根提供
+    :param after_commit: 数据提交后执行的消息、事件或上报编排
     :param kwargs: 订阅设置；owner_scope 为真时按用户名限定查重范围
     :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增
     """
@@ -145,12 +285,20 @@ def add_subscribe(
         return INCOMPLETE_IDENTITY
     identity, payload, username = translated
     oper = _get_subscribe_writer(subscribe_oper)
-    return oper.add(identity=identity, payload=payload, username=username)
+    if after_commit is None:
+        return oper.add(identity=identity, payload=payload, username=username)
+    return oper.add(
+        identity=identity,
+        payload=payload,
+        username=username,
+        after_commit=after_commit,
+    )
 
 
 async def async_add_subscribe(
     mediainfo: MediaInfo | MusicInfo,
     subscribe_oper: Optional[SubscribeWriter] = None,
+    after_commit: Optional[AsyncAfterCommitEffect] = None,
     **kwargs,
 ) -> Tuple[int, str]:
     """
@@ -158,6 +306,7 @@ async def async_add_subscribe(
 
     :param mediainfo: 识别结果
     :param subscribe_oper: 复用的订阅操作对象，未传时由启动组合根提供
+    :param after_commit: 数据提交后执行的异步消息、事件或上报编排
     :param kwargs: 订阅设置；owner_scope 为真时按用户名限定查重范围
     :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增
     """
@@ -166,12 +315,27 @@ async def async_add_subscribe(
         return INCOMPLETE_IDENTITY
     identity, payload, username = translated
     oper = _get_subscribe_writer(subscribe_oper)
-    return await oper.async_add(identity=identity, payload=payload, username=username)
+    if after_commit is None:
+        return await oper.async_add(identity=identity, payload=payload, username=username)
+    return await oper.async_add(
+        identity=identity,
+        payload=payload,
+        username=username,
+        after_commit=after_commit,
+    )
 
 
 __all__ = [
+    "AfterCommitEffect",
+    "AsyncCreateSubscriptionCommand",
+    "AsyncAfterCommitEffect",
+    "AsyncUnitOfWork",
+    "CreateSubscriptionCommand",
     "INCOMPLETE_IDENTITY",
+    "StagedSubscription",
+    "SubscriptionStagingRepository",
     "SubscribeWriter",
+    "UnitOfWork",
     "add_subscribe",
     "async_add_subscribe",
     "configure_subscribe_writer",

app/chain/subscribe.py

@@ -3,6 +3,7 @@ import json
 import random
 import threading
 import time
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Callable, Dict, List, Optional, Union, Tuple
 
@@ -88,6 +89,24 @@ SystemConfigOper = get_configured_system_config
 _DEFAULT_SYSTEM_CONFIG_PROVIDER = get_configured_system_config
 
 
+@dataclass(frozen=True, slots=True)
+class _SubscribePostCommitContext:
+    """订阅提交后副作用所需的不可变业务快照。"""
+
+    title: str
+    year: str
+    metainfo: MetaBase
+    mediainfo: MediaInfo
+    media_source: Optional[MediaSource]
+    media_id: Optional[str]
+    season: Optional[int]
+    channel: Optional[NotificationChannel]
+    source: Optional[str]
+    userid: Optional[str]
+    username: Optional[str]
+    message: bool
+
+
 def _system_config():
     """返回配置端口，并兼容旧测试对本地别名的替换。"""
     if SystemConfigOper is not _DEFAULT_SYSTEM_CONFIG_PROVIDER:
@@ -851,6 +870,98 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
             })
         return defaults
 
+    @staticmethod
+    def __subscribe_added_link(mtype: MediaType) -> str:
+        """返回订阅类型对应的前端详情入口。"""
+        if mtype == MediaType.TV:
+            return settings.MP_DOMAIN('#/subscribe/tv?tab=mysub')
+        if mtype == MediaType.MUSIC:
+            return settings.MP_DOMAIN('#/subscribe/music?tab=mysub')
+        return settings.MP_DOMAIN('#/subscribe/movie?tab=mysub')
+
+    @staticmethod
+    def __subscribe_report_payload(context: _SubscribePostCommitContext) -> dict:
+        """构造保持旧字段和值语义的订阅统计上报。"""
+        mediainfo = context.mediainfo
+        music_type = getattr(mediainfo, "music_type", None)
+        return {
+            "name": context.title,
+            "year": context.year,
+            "type": context.metainfo.type.value,
+            "media_source": context.media_source,
+            "media_id": context.media_id,
+            "music_type": music_type,
+            "total_tracks": getattr(mediainfo, "total_tracks", None)
+            if music_type == MUSIC_ENTITY_ALBUM else None,
+            "season": context.season,
+            "poster": mediainfo.get_poster_image(),
+            "backdrop": mediainfo.get_backdrop_image(),
+            "vote": mediainfo.vote_average,
+            "description": mediainfo.overview,
+        }
+
+    def __post_subscribe_added(
+        self,
+        subscribe_id: int,
+        context: _SubscribePostCommitContext,
+    ) -> None:
+        """同步执行提交后消息、事件和统计，异常不再触碰数据库事务。"""
+        if context.message:
+            self.post_message(
+                _SchemaMessage(
+                    channel=context.channel,
+                    source=context.source,
+                    mtype=MessageType.Subscribe,
+                    ctype=ContentType.SubscribeAdded,
+                    image=context.mediainfo.get_message_image(),
+                    link=self.__subscribe_added_link(context.mediainfo.type),
+                    userid=context.userid,
+                    username=context.username,
+                ),
+                meta=context.metainfo,
+                mediainfo=context.mediainfo,
+                username=context.username,
+            )
+        eventmanager.send_event(EventType.SubscribeAdded, {
+            "subscribe_id": subscribe_id,
+            "username": context.username,
+            "mediainfo": context.mediainfo.to_dict(),
+        })
+        MoviePilotServerHelper.sub_reg_async(
+            self.__subscribe_report_payload(context)
+        )
+
+    async def __async_post_subscribe_added(
+        self,
+        subscribe_id: int,
+        context: _SubscribePostCommitContext,
+    ) -> None:
+        """异步执行提交后消息、事件和统计，保持与同步入口相同顺序。"""
+        if context.message:
+            await self.async_post_message(
+                _SchemaMessage(
+                    channel=context.channel,
+                    source=context.source,
+                    mtype=MessageType.Subscribe,
+                    ctype=ContentType.SubscribeAdded,
+                    image=context.mediainfo.get_message_image(),
+                    link=self.__subscribe_added_link(context.mediainfo.type),
+                    userid=context.userid,
+                    username=context.username,
+                ),
+                meta=context.metainfo,
+                mediainfo=context.mediainfo,
+                username=context.username,
+            )
+        await eventmanager.async_send_event(EventType.SubscribeAdded, {
+            "subscribe_id": subscribe_id,
+            "username": context.username,
+            "mediainfo": context.mediainfo.to_dict(),
+        })
+        await MoviePilotServerHelper.async_sub_reg(
+            self.__subscribe_report_payload(context)
+        )
+
     def add(self, title: str, year: str,
             mtype: MediaType = None,
             episode_group: Optional[str] = None,
@@ -992,8 +1103,33 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
         # 添加订阅
         kwargs.update(self.__get_default_kwargs(mediainfo.type, **kwargs))
 
+        post_commit_context = _SubscribePostCommitContext(
+            title=title,
+            year=year,
+            metainfo=metainfo,
+            mediainfo=mediainfo,
+            media_source=media_source,
+            media_id=media_id,
+            season=season,
+            channel=channel,
+            source=source,
+            userid=userid,
+            username=username,
+            message=bool(message),
+        )
+
+        def _after_commit(subscribe_id: int) -> None:
+            """把同步提交后的副作用委托给单一顺序实现。"""
+            self.__post_subscribe_added(subscribe_id, post_commit_context)
+
         # 操作数据库
-        sid, err_msg = add_subscribe(mediainfo=mediainfo, season=season, username=username, **kwargs)
+        sid, err_msg = add_subscribe(
+            mediainfo=mediainfo,
+            season=season,
+            username=username,
+            after_commit=_after_commit,
+            **kwargs,
+        )
         if not sid:
             logger.error(f'{mediainfo.title_year} {err_msg}')
             if not exist_ok and message:
@@ -1007,51 +1143,6 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
                                                        image=mediainfo.get_message_image(),
                                                        userid=userid))
             return None, err_msg
-        elif message:
-            if mediainfo.type == MediaType.TV:
-                link = settings.MP_DOMAIN('#/subscribe/tv?tab=mysub')
-            elif mediainfo.type == MediaType.MUSIC:
-                link = settings.MP_DOMAIN('#/subscribe/music?tab=mysub')
-            else:
-                link = settings.MP_DOMAIN('#/subscribe/movie?tab=mysub')
-            # 订阅成功按规则发送消息
-            self.post_message(
-                _SchemaMessage(
-                    channel=channel,
-                    source=source,
-                    mtype=MessageType.Subscribe,
-                    ctype=ContentType.SubscribeAdded,
-                    image=mediainfo.get_message_image(),
-                    link=link,
-                    userid=userid,
-                    username=username
-                ),
-                meta=metainfo,
-                mediainfo=mediainfo,
-                username=username
-            )
-        # 发送事件
-        eventmanager.send_event(EventType.SubscribeAdded, {
-            "subscribe_id": sid,
-            "username": username,
-            "mediainfo": mediainfo.to_dict(),
-        })
-        # 统计订阅
-        MoviePilotServerHelper.sub_reg_async({
-            "name": title,
-            "year": year,
-            "type": metainfo.type.value,
-            "media_source": media_source,
-            "media_id": media_id,
-            "music_type": getattr(mediainfo, "music_type", None),
-            "total_tracks": getattr(mediainfo, "total_tracks", None)
-            if getattr(mediainfo, "music_type", None) == MUSIC_ENTITY_ALBUM else None,
-            "season": season,
-            "poster": mediainfo.get_poster_image(),
-            "backdrop": mediainfo.get_backdrop_image(),
-            "vote": mediainfo.vote_average,
-            "description": mediainfo.overview
-        })
         # 返回结果
         return sid, err_msg
 
@@ -1196,8 +1287,36 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
         # 列新默认参数
         kwargs.update(self.__get_default_kwargs(mediainfo.type, **kwargs))
 
+        post_commit_context = _SubscribePostCommitContext(
+            title=title,
+            year=year,
+            metainfo=metainfo,
+            mediainfo=mediainfo,
+            media_source=media_source,
+            media_id=media_id,
+            season=season,
+            channel=channel,
+            source=source,
+            userid=userid,
+            username=username,
+            message=bool(message),
+        )
+
+        async def _after_commit(subscribe_id: int) -> None:
+            """把异步提交后的副作用委托给单一顺序实现。"""
+            await self.__async_post_subscribe_added(
+                subscribe_id,
+                post_commit_context,
+            )
+
         # 操作数据库
-        sid, err_msg = await async_add_subscribe(mediainfo=mediainfo, season=season, username=username, **kwargs)
+        sid, err_msg = await async_add_subscribe(
+            mediainfo=mediainfo,
+            season=season,
+            username=username,
+            after_commit=_after_commit,
+            **kwargs,
+        )
         if not sid:
             logger.error(f'{mediainfo.title_year} {err_msg}')
             if not exist_ok and message:
@@ -1211,51 +1330,6 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
                                                                    image=mediainfo.get_message_image(),
                                                                    userid=userid))
             return None, err_msg
-        elif message:
-            if mediainfo.type == MediaType.TV:
-                link = settings.MP_DOMAIN('#/subscribe/tv?tab=mysub')
-            elif mediainfo.type == MediaType.MUSIC:
-                link = settings.MP_DOMAIN('#/subscribe/music?tab=mysub')
-            else:
-                link = settings.MP_DOMAIN('#/subscribe/movie?tab=mysub')
-            # 订阅成功按规则发送消息
-            await self.async_post_message(
-                _SchemaMessage(
-                    channel=channel,
-                    source=source,
-                    mtype=MessageType.Subscribe,
-                    ctype=ContentType.SubscribeAdded,
-                    image=mediainfo.get_message_image(),
-                    link=link,
-                    userid=userid,
-                    username=username
-                ),
-                meta=metainfo,
-                mediainfo=mediainfo,
-                username=username
-            )
-        # 发送事件
-        await eventmanager.async_send_event(EventType.SubscribeAdded, {
-            "subscribe_id": sid,
-            "username": username,
-            "mediainfo": mediainfo.to_dict(),
-        })
-        # 统计订阅
-        await MoviePilotServerHelper.async_sub_reg({
-            "name": title,
-            "year": year,
-            "type": metainfo.type.value,
-            "media_source": media_source,
-            "media_id": media_id,
-            "music_type": getattr(mediainfo, "music_type", None),
-            "total_tracks": getattr(mediainfo, "total_tracks", None)
-            if getattr(mediainfo, "music_type", None) == MUSIC_ENTITY_ALBUM else None,
-            "season": season,
-            "poster": mediainfo.get_poster_image(),
-            "backdrop": mediainfo.get_backdrop_image(),
-            "vote": mediainfo.vote_average,
-            "description": mediainfo.overview
-        })
         # 返回结果
         return sid, err_msg
 

app/db/oper/subscribe.py

@@ -8,9 +8,13 @@
 留在这一层的只有列类型强转与建库时间戳——它们跟着订阅表的列走，换谁来调都一样。
 """
 import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 from typing import Any, Tuple, List, Optional
 
-from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy import delete as sqlalchemy_delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
 
 from app.application.subscription.delete import SubscribeDeletionCandidate
 from app.db.base import DbOper
@@ -20,6 +24,18 @@ from app.schemas.types import MediaSource
 
 INTEGER_FLAG_FIELDS = ("best_version", "best_version_full", "search_imdbid", "manual_total_episode")
 
+AfterCommitEffect = Callable[[int], None]
+AsyncAfterCommitEffect = Callable[[int], Awaitable[None]]
+
+
+@dataclass(frozen=True, slots=True)
+class SubscribeStageResult:
+    """Oper 暂存结果，按 Application 端口需要暴露最小只读状态。"""
+
+    subscribe_id: int
+    message: str
+    created: bool
+
 
 def _normalize_integer_flags(payload: dict, fields: Tuple[str, ...] = INTEGER_FLAG_FIELDS) -> dict:
     """
@@ -67,6 +83,25 @@ class SubscribeOper(DbOper):
     订阅管理
     """
 
+    @staticmethod
+    def _identity_statement(identity: dict, username: Optional[str] = None):
+        """构造订阅查重语句，SQL 所有权收口在 Oper。"""
+        condition = Subscribe._identity_condition(  # pylint: disable=protected-access
+            identity.get("media_source"),
+            identity.get("media_id"),
+            identity.get("music_type"),
+        )
+        if condition is None or username == "":
+            return None
+        statement = select(Subscribe).where(condition)
+        if username:
+            statement = statement.where(Subscribe.username == username)
+        if identity.get("season") is not None:
+            statement = statement.where(Subscribe.season == identity["season"])
+        return statement.where(
+            Subscribe.episode_group == identity.get("episode_group")
+        )
+
     def _exists(self, identity: dict, username: Optional[str]) -> Optional[Any]:
         """
         按身份查重。
@@ -74,8 +109,18 @@ class SubscribeOper(DbOper):
         :param username: 非空时只在该用户的订阅内查
         :return: 命中的订阅行，未命中为 None
         """
+        if isinstance(self._db, Session):
+            statement = self._identity_statement(identity, username)
+            if statement is None:
+                return None
+            return self._db.execute(statement).scalars().first()
+        # 旧 SDK 允许无会话构造 Oper；保留其自动短会话行为，但规范入口不得走这里。
         if username:
-            return Subscribe.exists_by_username(self._db, username=username, **identity)
+            return Subscribe.exists_by_username(
+                self._db,
+                username=username,
+                **identity,
+            )
         return Subscribe.exists(self._db, **identity)
 
     async def _async_exists(self, identity: dict, username: Optional[str]) -> Optional[Any]:
@@ -85,12 +130,70 @@ class SubscribeOper(DbOper):
         :param username: 非空时只在该用户的订阅内查
         :return: 命中的订阅行，未命中为 None
         """
+        if isinstance(self._db, AsyncSession):
+            statement = self._identity_statement(identity, username)
+            if statement is None:
+                return None
+            result = await self._db.execute(statement)
+            return result.scalars().first()
+        # 同步路径一样只为无会话旧入口保留 Model 的自动短会话兼容。
         if username:
-            return await Subscribe.async_exists_by_username(self._db, username=username, **identity)
+            return await Subscribe.async_exists_by_username(
+                self._db,
+                username=username,
+                **identity,
+            )
         return await Subscribe.async_exists(self._db, **identity)
 
+    def stage_add(
+        self,
+        identity: dict,
+        payload: dict,
+        username: Optional[str] = None,
+    ) -> SubscribeStageResult:
+        """暂存同步新增并 flush 主键，不提交调用方拥有的事务。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("同步订阅新增需要调用方提供 Session")
+        subscribe = self._exists(identity, username)
+        if subscribe:
+            return SubscribeStageResult(
+                subscribe_id=subscribe.id,
+                message="订阅已存在",
+                created=False,
+            )
+        subscribe = Subscribe(**_persistable(payload))
+        self._db.add(subscribe)
+        self._db.flush()
+        if not subscribe.id:
+            return SubscribeStageResult(0, "新增订阅失败", True)
+        return SubscribeStageResult(subscribe.id, "新增订阅成功", True)
+
+    async def async_stage_add(
+        self,
+        identity: dict,
+        payload: dict,
+        username: Optional[str] = None,
+    ) -> SubscribeStageResult:
+        """暂存异步新增并 flush 主键，不提交调用方拥有的事务。"""
+        if not isinstance(self._db, AsyncSession):
+            raise RuntimeError("异步订阅新增需要调用方提供 AsyncSession")
+        subscribe = await self._async_exists(identity, username)
+        if subscribe:
+            return SubscribeStageResult(
+                subscribe_id=subscribe.id,
+                message="订阅已存在",
+                created=False,
+            )
+        subscribe = Subscribe(**_persistable(payload))
+        self._db.add(subscribe)
+        await self._db.flush()
+        if not subscribe.id:
+            return SubscribeStageResult(0, "新增订阅失败", True)
+        return SubscribeStageResult(subscribe.id, "新增订阅成功", True)
+
     def add(self, identity: dict, payload: dict,
-            username: Optional[str] = None) -> Tuple[int, str]:
+            username: Optional[str] = None,
+            after_commit: Optional[AfterCommitEffect] = None) -> Tuple[int, str]:
         """
         新增订阅：命中既有订阅则原样返回，否则落库后回读。
 
@@ -99,33 +202,44 @@ class SubscribeOper(DbOper):
         :param identity: 查重身份（media_source/media_id/music_type/season/episode_group）
         :param payload: 订阅表的写入字段，媒体翻译由 application/subscription/write.py 完成
         :param username: 非空时把查重限定在该用户的订阅内
+        :param after_commit: 兼容旧调用方的提交后副作用；新入口由 Application Command 调用
         :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增
         """
         subscribe = self._exists(identity, username)
         if subscribe:
+            if after_commit:
+                after_commit(subscribe.id)
             return subscribe.id, "订阅已存在"
         Subscribe(**_persistable(payload)).create(self._db)
         subscribe = self._exists(identity, username)
         if not subscribe:
             return 0, "新增订阅失败"
+        if after_commit:
+            after_commit(subscribe.id)
         return subscribe.id, "新增订阅成功"
 
     async def async_add(self, identity: dict, payload: dict,
-                        username: Optional[str] = None) -> Tuple[int, str]:
+                        username: Optional[str] = None,
+                        after_commit: Optional[AsyncAfterCommitEffect] = None) -> Tuple[int, str]:
         """
         异步新增订阅，语义与 add 完全一致。
         :param identity: 查重身份（media_source/media_id/music_type/season/episode_group）
         :param payload: 订阅表的写入字段，媒体翻译由 application/subscription/write.py 完成
         :param username: 非空时把查重限定在该用户的订阅内
+        :param after_commit: 兼容旧调用方的异步提交后副作用
         :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增
         """
         subscribe = await self._async_exists(identity, username)
         if subscribe:
+            if after_commit:
+                await after_commit(subscribe.id)
             return subscribe.id, "订阅已存在"
         await Subscribe(**_persistable(payload)).async_create(self._db)
         subscribe = await self._async_exists(identity, username)
         if not subscribe:
             return 0, "新增订阅失败"
+        if after_commit:
+            await after_commit(subscribe.id)
         return subscribe.id, "新增订阅成功"
 
     def exists(
@@ -143,7 +257,7 @@ class SubscribeOper(DbOper):
             "season": season,
             "episode_group": episode_group,
         }
-        return bool(Subscribe.exists(self._db, **identity_params))
+        return bool(self._exists(identity_params, username=None))
 
     async def async_exists(
             self, media_source: MediaSource, media_id: str,
@@ -151,13 +265,15 @@ class SubscribeOper(DbOper):
             music_type: Optional[str] = None,
     ) -> Optional[Subscribe]:
         """异步按媒体身份、季号及可选剧集组读取命中的订阅。"""
-        return await Subscribe.async_exists(
+        return await self._async_exists(
-            self._db,
+            {
-            media_source=media_source,
+                "media_source": media_source,
-            media_id=media_id,
+                "media_id": media_id,
-            music_type=music_type,
+                "music_type": music_type,
-            season=season,
+                "season": season,
-            episode_group=episode_group,
+                "episode_group": episode_group,
+            },
+            username=None,
         )
 
     def get(self, sid: int) -> Optional[Subscribe]:

app/startup/modules_initializer.py

@@ -61,7 +61,13 @@ from app.adapters.external.server import (
 )
 from app.application.server.report import ServerReportService
 from app.application.server.share import ServerSharingService
-from app.db.session import close_database, get_async_db, get_db
+from app.db.session import (
+    SessionFactory,
+    async_session_scope,
+    close_database,
+    get_async_db,
+    get_db,
+)
 from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
 from app.db.oper.subscribe import SubscribeOper
 from app.db.oper.agentchat import AgentChatOper
@@ -90,6 +96,7 @@ from app.startup.managed_resources_initializer import (
     init_managed_resources,
     stop_managed_resources,
 )
+from app.startup.subscription import TransactionalSubscribeWriter
 from app.adapters.web.security.access import set_superuser_token_payload_provider
 from app.application.security.auth import build_superuser_token_payload
 from app.application.image import configure_wallpaper_providers
@@ -465,7 +472,12 @@ async def init_modules():
         workflow=lambda: WorkflowOper(),
         plugin_data=lambda: PluginDataOper(),
     )
-    configure_subscribe_writer(lambda: SubscribeOper())
+    configure_subscribe_writer(
+        lambda: TransactionalSubscribeWriter(
+            sync_session=SessionFactory,
+            async_session=async_session_scope,
+        )
+    )
     # 托管资源只在这里装配声明与 adapter，具体资源仍由首个消费者显式激活。
     init_managed_resources()
     # 应用服务不反向依赖 Chain，由启动组合层注入壁纸来源。

app/startup/subscription.py

@@ -0,0 +1,70 @@
+"""订阅写入事务适配器的启动装配。"""
+
+from collections.abc import Callable
+from contextlib import AbstractAsyncContextManager
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
+
+from app.application.subscription.write import (
+    AfterCommitEffect,
+    AsyncAfterCommitEffect,
+    AsyncCreateSubscriptionCommand,
+    CreateSubscriptionCommand,
+)
+from app.db.oper.subscribe import SubscribeOper
+from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
+
+
+class TransactionalSubscribeWriter:
+    """为每次订阅新增创建独占会话，并把提交权交给 Application Command。"""
+
+    def __init__(
+        self,
+        sync_session: Callable[[], Session],
+        async_session: Callable[
+            [],
+            AbstractAsyncContextManager[AsyncSession],
+        ],
+    ) -> None:
+        """注入同步会话工厂和异步会话作用域。"""
+        self._sync_session = sync_session
+        self._async_session = async_session
+
+    def add(
+        self,
+        identity: dict,
+        payload: dict,
+        username: str | None = None,
+        after_commit: AfterCommitEffect | None = None,
+    ) -> tuple[int, str]:
+        """在独占同步会话内执行一次完整订阅新增事务。"""
+        session = self._sync_session()
+        try:
+            command = CreateSubscriptionCommand(
+                repository=SubscribeOper(session),
+                unit_of_work=SqlAlchemyUnitOfWork(session),
+            )
+            return command.execute(identity, payload, username, after_commit)
+        finally:
+            session.close()
+
+    async def async_add(
+        self,
+        identity: dict,
+        payload: dict,
+        username: str | None = None,
+        after_commit: AsyncAfterCommitEffect | None = None,
+    ) -> tuple[int, str]:
+        """在独占异步会话作用域内执行一次完整订阅新增事务。"""
+        async with self._async_session() as session:
+            command = AsyncCreateSubscriptionCommand(
+                repository=SubscribeOper(session),
+                unit_of_work=SqlAlchemyAsyncUnitOfWork(session),
+            )
+            return await command.execute(
+                identity,
+                payload,
+                username,
+                after_commit,
+            )

docs/architecture-overview.md

@@ -361,8 +361,11 @@ flowchart LR
   归 `app/application/`（见 `application/subscription/write.py`、`application/history.py`）。
   订阅新增、查询、变更、删除、身份和搜索契约已经统一收口在 `application/subscription/`，
   不再保留主题包之外的第二个写入入口。
-- Oper 只 stage mutation，不创建独立 Session、不提交；Application Command 通过请求或任务
+- 规范写入口中的 Oper 只 stage mutation，不创建独立 Session、不提交；Application Command
-  入口注入的 UnitOfWork 统一 `commit/rollback`，事件、刷新和上报只在 commit 成功后执行。
+  通过请求或任务入口注入的 UnitOfWork 统一 `commit/rollback`，事件、刷新和上报只在 commit
+  成功后执行。订阅新增样板由 `startup/subscription.py` 创建独占 Session，
+  `application/subscription/write.py` 决定事务与 post-commit 边界，`SubscribeOper.stage_add()`
+  只查重、`add` 和 `flush`。旧 SDK 显式构造的无会话 Oper 暂留兼容自动短会话，不得被新代码复用。
   `transaction-debt-baseline.json` 将存量 178 个 Model 事务装饰器冻结为只降不增低水位。
 - 每次表结构变更必须新增 `database/versions/` 下的 Alembic 迁移。
 - 运行期业务配置使用 `SystemConfigKey` 枚举 + `SystemConfigOper`，禁止裸字符串键；

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 已完成，后续任务按 ID 独立提交和回滚
+> 实施进度：阶段 0（ARCH-201～203）、阶段 1（ARCH-210～212）与 ARCH-220～221 已完成，后续任务按 ID 独立提交和回滚
 
 ## 1. 结论先行
 
@@ -373,6 +373,20 @@ flowchart TB
 
 **完成标准**：一个业务动作只有一个事务所有者；任意入口都不会因内部 Model 方法提前 commit 而产生部分写入。
 
+**实施记录（2026-08-21）**：
+
+- `app/startup/subscription.py` 为每次规范新增创建独占同步/异步 Session；
+  `CreateSubscriptionCommand` / `AsyncCreateSubscriptionCommand` 持有 UoW，Oper 只执行
+  查重、`add` 与 `flush`。
+- `SubscribeOper.stage_add()` 的查重 SQL 已收口到 Oper，不再调用 Model 自动会话装饰器；
+  无会话构造 `SubscribeOper()` 的旧 SDK 路径保留原自动短会话和返回值，未扩散为规范入口。
+- Chain 把原有“成功消息 → `SubscribeAdded` 事件 → Server 统计”作为显式 post-commit
+  回调交给 Command；commit/flush 失败回滚，事件或上报失败只传播原异常，不回滚已提交记录。
+- 同步/异步 `SubscribeChain.add` 方法长度从各 203 行降至 183/186 行；新增 9 个事务边界测试，
+  覆盖成功顺序、commit/flush 失败、重复请求、Oper 不提交、事件失败、上报失败与真实落库。
+- Model 装饰器总数仍为 178：本切片绕开了继承自 `Base.create/async_create` 的自动提交，
+  但为保留既有 Model/旧 SDK 查询兼容未机械删除查询装饰器；ratchet 保持不增，后续切片继续下降。
+
 #### ARCH-222：按风险迁移其余写用例
 
 推荐顺序：

docs/rules/10-data-and-persistent.md

@@ -99,6 +99,11 @@ Oper classes accept and return persistence values. Turning a `MediaInfo` or
 - A synchronous Session is private to one worker thread. An AsyncSession is
   private to one asyncio task/operation; neither may be stored in a process
   singleton or reused by concurrent work.
+- Subscription creation is the reference slice: `app/startup/subscription.py`
+  creates an exclusive Session, `app/application/subscription/write.py` owns the
+  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.
 
 Run `./.venv/bin/python scripts/architecture/baseline.py --check-host` after
 persistence changes. A deliberate debt reduction may refresh the low-water mark
@@ -214,4 +219,4 @@ When `REDIS_HOST` is configured, `app/modules/redis/` provides a distributed cac
 - `settings.API_TOKEN` and other secret fields must not be included in log output or API responses.
 - The `config list --show-secrets` flag exists specifically to gate secret visibility in the CLI.
 
-*Last Updated: 2026-08-14*
+*Last Updated: 2026-08-21*

tests/conftest.py

@@ -30,13 +30,19 @@ def configure_plugin_system_services():
     from app.api.data import configure_api_data_ports
     from app.application.configuration import SystemConfigService, configure_system_config
     from app.application.service import configure_service_directory
-    from app.db.session import get_async_db, get_db
+    from app.db.session import (
+        SessionFactory,
+        async_session_scope,
+        get_async_db,
+        get_db,
+    )
     from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
     from app.db.oper.systemconfig import SystemConfigOper
 
     configure_token_codec(create_access_token, decode_access_token)
     configure_system_config(SystemConfigService(repository=SystemConfigOper()))
     from app.application.chain.data import configure_chain_data_ports
+    from app.application.subscription.write import configure_subscribe_writer
     from app.application.plugin.runtime import configure_plugin_runtime
     from app.application.module import configure_module_runtime
     from app.application.chain.context import (
@@ -73,6 +79,7 @@ def configure_plugin_system_services():
     from app.db.oper.workflow import WorkflowOper
     from app.db.oper.message import MessageOper
     from app.db.oper.passkey import PassKeyOper
+    from app.startup.subscription import TransactionalSubscribeWriter
 
     configure_api_data_ports(
         sync_session=get_db,
@@ -100,6 +107,12 @@ def configure_plugin_system_services():
             "sync": SqlAlchemyUnitOfWork,
         },
     )
+    configure_subscribe_writer(
+        lambda: TransactionalSubscribeWriter(
+            sync_session=SessionFactory,
+            async_session=async_session_scope,
+        )
+    )
 
     configure_chain_data_ports(
         site=lambda: SiteOper(),

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6084,
+  "edge_count": 6092,
-  "edge_sha256": "0dbce22b56284b845591fd38771a96c937ec898db7e9d1aece1ec47ceff8de04",
+  "edge_sha256": "1935d43d8d3c0c3e3687109f0b81a4e56cbc4bc2ed4763a706e2b0d7c119f378",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -5873,6 +5873,7 @@
     "app.startup.modules_initializer -> app.startup.agent_initializer",
     "app.startup.modules_initializer -> app.startup.database",
     "app.startup.modules_initializer -> app.startup.managed_resources_initializer",
+    "app.startup.modules_initializer -> app.startup.subscription",
     "app.startup.monitor_initializer -> app.monitor",
     "app.startup.plugins_initializer -> app.adapters",
     "app.startup.plugins_initializer -> app.adapters.external",
@@ -5922,6 +5923,13 @@
     "app.startup.scheduler_initializer -> app.application",
     "app.startup.scheduler_initializer -> app.application.scheduling",
     "app.startup.scheduler_initializer -> app.scheduler",
+    "app.startup.subscription -> app.application",
+    "app.startup.subscription -> app.application.subscription",
+    "app.startup.subscription -> app.application.subscription.write",
+    "app.startup.subscription -> app.db",
+    "app.startup.subscription -> app.db.oper",
+    "app.startup.subscription -> app.db.oper.subscribe",
+    "app.startup.subscription -> app.db.uow",
     "app.startup.transfer_initializer -> app.chain",
     "app.startup.transfer_initializer -> app.chain.transfer",
     "app.startup.workflow_initializer -> app.workflow",
@@ -6101,7 +6109,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 756,
+  "module_count": 757,
   "modules": [
     "app",
     "app.adapters",
@@ -6836,6 +6844,7 @@
     "app.startup.plugins_initializer",
     "app.startup.routers_initializer",
     "app.startup.scheduler_initializer",
+    "app.startup.subscription",
     "app.startup.transfer_initializer",
     "app.startup.workflow_initializer",
     "app.testing",

tests/test_subscribe_create_command.py

@@ -0,0 +1,240 @@
+"""订阅新增事务所有权与默认入口集成测试。"""
+
+import asyncio
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from app.application.subscription.write import (
+    AsyncCreateSubscriptionCommand,
+    CreateSubscriptionCommand,
+    add_subscribe,
+    async_add_subscribe,
+)
+from app.db.models.subscribe import Subscribe
+from app.db.oper.subscribe import SubscribeOper, SubscribeStageResult
+from app.domain.context import MediaInfo
+from app.schemas.types import MediaSource, MediaType
+
+
+def _media(media_id: str) -> MediaInfo:
+    """构造默认事务写入路径所需的最小媒体信息。"""
+    media = MediaInfo()
+    media.type = MediaType.MOVIE
+    media.title = "事务测试电影"
+    media.year = "2026"
+    media.media_source = MediaSource.TMDB
+    media.media_id = media_id
+    media.vote_average = 8.0
+    media.overview = "事务切片"
+    return media
+
+
+def test_sync_command_orders_stage_commit_before_caller_effect() -> None:
+    """同步新增只有在仓储暂存和提交成功后才把结果交给外部副作用。"""
+    calls: list[str] = []
+    repository = Mock()
+    repository.stage_add.side_effect = lambda *_: (
+        calls.append("stage")
+        or SubscribeStageResult(10, "新增订阅成功", True)
+    )
+    unit_of_work = Mock()
+    unit_of_work.commit.side_effect = lambda: calls.append("commit")
+    command = CreateSubscriptionCommand(repository, unit_of_work)
+
+    result = command.execute({"media_id": "10"}, {"name": "demo"})
+    calls.append("effect")
+
+    assert result == (10, "新增订阅成功")
+    assert calls == ["stage", "commit", "effect"]
+    unit_of_work.rollback.assert_not_called()
+
+
+def test_sync_command_rolls_back_commit_failure_and_skips_effect() -> None:
+    """提交失败必须回滚并传播原异常，调用方不能误执行提交后副作用。"""
+    commit_error = RuntimeError("commit failed")
+    repository = Mock()
+    repository.stage_add.return_value = SubscribeStageResult(
+        11,
+        "新增订阅成功",
+        True,
+    )
+    unit_of_work = Mock()
+    unit_of_work.commit.side_effect = commit_error
+    command = CreateSubscriptionCommand(repository, unit_of_work)
+    effects: list[str] = []
+
+    with pytest.raises(RuntimeError) as raised:
+        command.execute({"media_id": "11"}, {"name": "demo"})
+        effects.append("effect")
+
+    assert raised.value is commit_error
+    assert effects == []
+    unit_of_work.rollback.assert_called_once_with()
+
+
+def test_sync_command_does_not_commit_duplicate_request() -> None:
+    """查重命中沿用旧 ID 和消息，不开启无意义写事务。"""
+    repository = Mock()
+    repository.stage_add.return_value = SubscribeStageResult(
+        12,
+        "订阅已存在",
+        False,
+    )
+    unit_of_work = Mock()
+    command = CreateSubscriptionCommand(repository, unit_of_work)
+
+    assert command.execute({}, {}) == (12, "订阅已存在")
+    unit_of_work.commit.assert_not_called()
+    unit_of_work.rollback.assert_not_called()
+
+
+def test_sync_event_failure_does_not_roll_back_committed_subscription() -> None:
+    """事件属于提交后副作用，失败只向上传播且不能伪装成数据库回滚。"""
+    calls: list[str] = []
+    repository = Mock()
+    repository.stage_add.return_value = SubscribeStageResult(
+        13,
+        "新增订阅成功",
+        True,
+    )
+    unit_of_work = Mock()
+    unit_of_work.commit.side_effect = lambda: calls.append("commit")
+    event_error = RuntimeError("event failed")
+
+    def send_event(_subscribe_id: int) -> None:
+        """模拟 Chain 在提交后发送订阅事件失败。"""
+        calls.append("event")
+        raise event_error
+
+    command = CreateSubscriptionCommand(repository, unit_of_work)
+
+    with pytest.raises(RuntimeError) as raised:
+        command.execute({}, {}, after_commit=send_event)
+
+    assert raised.value is event_error
+    assert calls == ["commit", "event"]
+    unit_of_work.rollback.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_async_command_rolls_back_staging_failure() -> None:
+    """异步 flush 或唯一约束失败同样由命令回滚，不留部分写入。"""
+    staging_error = RuntimeError("flush failed")
+    repository = Mock()
+    repository.async_stage_add = AsyncMock(side_effect=staging_error)
+    unit_of_work = Mock()
+    unit_of_work.commit = AsyncMock()
+    unit_of_work.rollback = AsyncMock()
+    command = AsyncCreateSubscriptionCommand(repository, unit_of_work)
+
+    with pytest.raises(RuntimeError) as raised:
+        await command.execute({}, {})
+
+    assert raised.value is staging_error
+    unit_of_work.commit.assert_not_awaited()
+    unit_of_work.rollback.assert_awaited_once_with()
+
+
+@pytest.mark.asyncio
+async def test_async_report_failure_happens_after_event_without_rollback() -> None:
+    """异步上报失败保留事件先行顺序，也不回滚已经提交的订阅。"""
+    calls: list[str] = []
+    repository = Mock()
+    repository.async_stage_add = AsyncMock(
+        return_value=SubscribeStageResult(14, "新增订阅成功", True)
+    )
+    unit_of_work = Mock()
+    unit_of_work.commit = AsyncMock(side_effect=lambda: calls.append("commit"))
+    unit_of_work.rollback = AsyncMock()
+    report_error = RuntimeError("report failed")
+
+    async def send_event_and_report(_subscribe_id: int) -> None:
+        """模拟 Chain 先发事件再执行统计上报。"""
+        calls.append("event")
+        calls.append("report")
+        raise report_error
+
+    command = AsyncCreateSubscriptionCommand(repository, unit_of_work)
+
+    with pytest.raises(RuntimeError) as raised:
+        await command.execute({}, {}, after_commit=send_event_and_report)
+
+    assert raised.value is report_error
+    assert calls == ["commit", "event", "report"]
+    unit_of_work.rollback.assert_not_awaited()
+
+
+def test_default_sync_writer_persists_once_and_reuses_duplicate(db) -> None:
+    """Chain 默认入口使用独立事务写入，重复媒体身份返回同一订阅。"""
+    db.watermark(Subscribe)
+    media = _media("arch-221-sync")
+    after_commit = Mock()
+
+    first = add_subscribe(mediainfo=media, after_commit=after_commit)
+    second = add_subscribe(mediainfo=media, after_commit=after_commit)
+
+    assert first[0] > 0
+    assert first[1] == "新增订阅成功"
+    assert second == (first[0], "订阅已存在")
+    db.session.expire_all()
+    rows = Subscribe.list_by_media_identity(
+        db.session,
+        media_source=MediaSource.TMDB,
+        media_id="arch-221-sync",
+    )
+    assert [row.id for row in rows] == [first[0]]
+    assert after_commit.call_args_list == [
+        ((first[0],), {}),
+        ((first[0],), {}),
+    ]
+
+
+def test_stage_add_executes_identity_sql_in_oper(db, monkeypatch) -> None:
+    """规范新增路径直接由 Oper 查询，不能退回 Model 自动会话装饰器。"""
+    db.watermark(Subscribe)
+    commit = Mock(wraps=db.session.commit)
+    monkeypatch.setattr(db.session, "commit", commit)
+    monkeypatch.setattr(
+        Subscribe,
+        "exists",
+        Mock(side_effect=AssertionError("model query must not run")),
+    )
+    oper = SubscribeOper(db.session)
+    identity = {
+        "media_source": str(MediaSource.TMDB),
+        "media_id": "arch-221-stage",
+        "music_type": None,
+        "season": None,
+        "episode_group": None,
+    }
+
+    staged = oper.stage_add(
+        identity,
+        {
+            "name": "Oper SQL",
+            "type": MediaType.MOVIE.value,
+            "state": "N",
+            **identity,
+        },
+    )
+
+    assert staged.created is True
+    assert staged.subscribe_id > 0
+    commit.assert_not_called()
+    db.session.rollback()
+
+
+def test_default_async_writer_persists_committed_row(db) -> None:
+    """Agent/API 使用的异步 Chain 入口在返回前已完成请求级提交。"""
+    db.watermark(Subscribe)
+    media = _media("arch-221-async")
+
+    subscribe_id, message = asyncio.run(async_add_subscribe(mediainfo=media))
+
+    assert subscribe_id > 0
+    assert message == "新增订阅成功"
+    db.session.expire_all()
+    persisted = Subscribe.get(db.session, subscribe_id)
+    assert persisted is not None
+    assert persisted.media_id == "arch-221-async"