refactor: unify subscription deletion transactions

app/application/messaging/subscribe.py

@@ -23,10 +23,18 @@ class SubscribeInteractionActions(Protocol):
     声明订阅交互需要调用的业务动作。
     """
 
-    def refresh(self):
+    def refresh(self) -> Any:
         """执行订阅刷新。"""
         ...
 
+    def check(self) -> Any:
+        """执行订阅元数据检查。"""
+        ...
+
+    def search(self, **kwargs: Any) -> Any:
+        """按消息入口参数执行订阅搜索。"""
+        ...
+
 
 class SubscribeInteractionRepository(Protocol):
     """订阅消息交互所需的同步数据端口。"""
@@ -37,14 +45,11 @@ class SubscribeInteractionRepository(Protocol):
     def get(self, subscribe_id: int) -> Optional[Any]:
         """按 ID 返回订阅。"""
 
-    def delete(self, subscribe_id: int) -> Any:
-        """删除订阅。"""
-
-    def check(self):
+    def check(self) -> Any:
         """执行订阅元数据检查。"""
         ...
 
-    def search(self, **kwargs):
+    def search(self, **kwargs: Any) -> Any:
         """执行订阅搜索。"""
         ...
 
@@ -62,15 +67,15 @@ class SubscribeInteractionHandler:
             messenger: MessageGateway,
             actions: SubscribeInteractionActions,
             repository: SubscribeInteractionRepository,
-            report_deleted: Callable[[dict], Any],
-    ):
+            delete_subscription: Callable[[int], bool],
+    ) -> None:
         """
         注入消息投递接口和订阅业务动作。
         """
         self._messenger = messenger
         self._actions = actions
         self._repository = repository
-        self._report_deleted = report_deleted
+        self._delete_subscription = delete_subscription
 
     def remote_list(
             self,
@@ -727,15 +732,10 @@ class SubscribeInteractionHandler:
             if not subscribe:
                 missing.append(str(subscribe_id))
                 continue
+            if not self._delete_subscription(subscribe_id):
+                missing.append(str(subscribe_id))
+                continue
             deleted.append(subscribe.name)
-            self._repository.delete(subscribe_id)
-            self._report_deleted(
-                {
-                    "media_source": subscribe.media_source,
-                    "media_id": subscribe.media_id,
-                    "season": subscribe.season,
-                }
-            )
 
         if not deleted and missing:
             return False, f"未找到订阅：{', '.join(missing)}"

app/application/subscription/delete.py

@@ -1,13 +1,18 @@
 """订阅删除应用用例及其依赖端口。"""
 
-from contextlib import AbstractAsyncContextManager
+from contextlib import AbstractAsyncContextManager, AbstractContextManager
 from dataclasses import dataclass
 from datetime import datetime, timezone
 import inspect
 from typing import Any, Awaitable, Callable, Mapping, Protocol, cast
 from uuid import uuid4
 
-from app.application.outbox import AsyncOutboxTransaction, OutboxIntent
+from app.application.outbox import (
+    AsyncOutboxTransaction,
+    OutboxIntent,
+    SyncOutboxTransaction,
+    SyncUnitOfWork,
+)
 from app.schemas.event import SubscribeDeletedEventData
 
 
@@ -43,6 +48,20 @@ class SubscribeDeletionRepository(Protocol):
         ...
 
 
+class SyncSubscribeDeletionRepository(Protocol):
+    """同步消息入口执行订阅删除所需的最小数据访问端口。"""
+
+    def get_candidate_sync(
+        self,
+        subscribe_id: int,
+    ) -> SubscribeDeletionCandidate | None:
+        """读取订阅及删除事件所需的稳定快照。"""
+        ...
+
+    def stage_delete_sync(self, subscribe_id: int) -> None:
+        """把已读取的订阅登记为待删除，但不自行提交事务。"""
+        ...
+
 class AsyncUnitOfWork(Protocol):
     """订阅写用例使用的异步事务端口。"""
 
@@ -57,6 +76,18 @@ class AsyncUnitOfWork(Protocol):
 
 SubscribeDeletedPublisher = Callable[[dict[str, Any]], Awaitable[None]]
 SubscribeDeletedReporter = Callable[[Mapping[str, object]], object | Awaitable[object]]
+SyncSubscribeDeletedPublisher = Callable[[dict[str, Any]], None]
+SyncSubscribeDeletedReporter = Callable[[Mapping[str, object]], object]
+
+
+@dataclass(frozen=True)
+class _SubscribeDeletionEffects:
+    """同步和异步入口共用的删除事件、统计与 outbox 意图。"""
+
+    event_payload: dict[str, Any]
+    report_payload: dict[str, object]
+    event_intent: OutboxIntent
+    report_intent: OutboxIntent
 
 
 class DeleteSubscribeCommand:
@@ -89,36 +120,23 @@ class DeleteSubscribeCommand:
         提交后的事件与上报保持原有顺序，任一副作用失败都会继续向调用方抛出。
         """
         candidate = await self._repository.get_candidate(subscribe_id)
-        if not self._can_delete(candidate, actor):
+        if not can_delete_subscribe(candidate, actor):
             return False
         assert candidate is not None
 
-        await self._repository.stage_delete(subscribe_id)
-        event_payload = build_subscribe_deleted_payload(
+        effects = _build_deletion_effects(
             subscribe_id,
             candidate.event_payload,
         )
-        event_key = event_payload["idempotency_key"]
-        report_key = f"{event_key}:report"
         try:
+            await self._repository.stage_delete(subscribe_id)
             if self._outbox:
                 await self._outbox.stage(
-                    OutboxIntent(
-                        event_key=event_key,
-                        topic="subscribe.deleted",
-                        payload=event_payload,
-                    ),
+                    effects.event_intent,
                     datetime.now(timezone.utc),
                 )
                 await self._outbox.stage(
-                    OutboxIntent(
-                        event_key=report_key,
-                        topic="subscribe.deleted.report",
-                        payload={
-                            "idempotency_key": report_key,
-                            "subscribe_info": dict(candidate.event_payload),
-                        },
-                    ),
+                    effects.report_intent,
                     datetime.now(timezone.utc),
                 )
             await self._unit_of_work.commit()
@@ -126,37 +144,125 @@ class DeleteSubscribeCommand:
             await self._unit_of_work.rollback()
             raise
 
-        await self._publish_deleted(event_payload)
+        await self._publish_deleted(effects.event_payload)
         if self._outbox:
             await self._outbox.complete_by_event_key(
-                event_key,
+                effects.event_intent.event_key,
                 datetime.now(timezone.utc),
             )
         # 上报适配器会自行白名单过滤公开字段；传完整删除前快照可保留音乐实体维度，
         # 避免 Agent 与 API 入口收敛后丢失 music_type / total_tracks。
-        report_result = self._report_deleted(dict(candidate.event_payload))
+        report_result = self._report_deleted(effects.report_payload)
         if inspect.isawaitable(report_result):
             report_result = await report_result
         if report_result is False:
             raise RuntimeError("订阅删除统计上报未确认")
         if self._outbox:
             await self._outbox.complete_by_event_key(
-                report_key,
+                effects.report_intent.event_key,
                 datetime.now(timezone.utc),
             )
         return True
 
-    @staticmethod
-    def _can_delete(
-        candidate: SubscribeDeletionCandidate | None,
+
+class SyncDeleteSubscribeCommand:
+    """为同步消息入口执行同一订阅删除事务与 durable 副作用协议。"""
+
+    def __init__(
+        self,
+        repository: SyncSubscribeDeletionRepository,
+        unit_of_work: SyncUnitOfWork,
+        publish_deleted: SyncSubscribeDeletedPublisher,
+        report_deleted: SyncSubscribeDeletedReporter,
+        outbox: SyncOutboxTransaction | None = None,
+    ) -> None:
+        """注入同步数据访问、事务与提交后副作用端口。"""
+        self._repository = repository
+        self._unit_of_work = unit_of_work
+        self._publish_deleted = publish_deleted
+        self._report_deleted = report_deleted
+        self._outbox = outbox
+
+    def execute(
+        self,
+        subscribe_id: int,
         actor: SubscribeDeletionActor,
     ) -> bool:
-        """判断用户是否拥有目标订阅的删除权限。"""
-        if candidate is None:
+        """同步删除当前用户可访问的订阅，并保持事件和统计的可靠投递顺序。"""
+        candidate = self._repository.get_candidate_sync(subscribe_id)
+        if not can_delete_subscribe(candidate, actor):
             return False
-        if actor.is_superuser:
-            return True
-        return bool(candidate.username) and candidate.username == actor.username
+        assert candidate is not None
+
+        effects = _build_deletion_effects(
+            subscribe_id,
+            candidate.event_payload,
+        )
+        try:
+            self._repository.stage_delete_sync(subscribe_id)
+            if self._outbox:
+                now = datetime.now(timezone.utc)
+                self._outbox.stage(effects.event_intent, now)
+                self._outbox.stage(effects.report_intent, now)
+            self._unit_of_work.commit()
+        except Exception:
+            self._unit_of_work.rollback()
+            raise
+
+        self._publish_deleted(effects.event_payload)
+        if self._outbox:
+            self._outbox.complete_by_event_key(
+                effects.event_intent.event_key,
+                datetime.now(timezone.utc),
+            )
+        if self._report_deleted(effects.report_payload) is False:
+            raise RuntimeError("订阅删除统计上报未确认")
+        if self._outbox:
+            self._outbox.complete_by_event_key(
+                effects.report_intent.event_key,
+                datetime.now(timezone.utc),
+            )
+        return True
+
+
+def can_delete_subscribe(
+    candidate: SubscribeDeletionCandidate | None,
+    actor: SubscribeDeletionActor,
+) -> bool:
+    """判断用户是否拥有目标订阅的删除权限。"""
+    if candidate is None:
+        return False
+    if actor.is_superuser:
+        return True
+    return bool(candidate.username) and candidate.username == actor.username
+
+
+def _build_deletion_effects(
+    subscribe_id: int,
+    subscribe_info: Mapping[str, object],
+) -> _SubscribeDeletionEffects:
+    """一次性构造两种执行风格共用的事件、上报和 durable intent。"""
+    event_payload = build_subscribe_deleted_payload(subscribe_id, subscribe_info)
+    event_key = event_payload["idempotency_key"]
+    report_key = f"{event_key}:report"
+    report_payload = dict(subscribe_info)
+    return _SubscribeDeletionEffects(
+        event_payload=event_payload,
+        report_payload=report_payload,
+        event_intent=OutboxIntent(
+            event_key=event_key,
+            topic="subscribe.deleted",
+            payload=event_payload,
+        ),
+        report_intent=OutboxIntent(
+            event_key=report_key,
+            topic="subscribe.deleted.report",
+            payload={
+                "idempotency_key": report_key,
+                "subscribe_info": report_payload,
+            },
+        ),
+    )
 
 
 def build_subscribe_deleted_payload(
@@ -176,7 +282,9 @@ def build_subscribe_deleted_payload(
 
 
 DeleteSubscribeScope = Callable[[], AbstractAsyncContextManager[DeleteSubscribeCommand]]
+SyncDeleteSubscribeScope = Callable[[], AbstractContextManager[SyncDeleteSubscribeCommand]]
 _configured_delete_scope: DeleteSubscribeScope | None = None
+_configured_sync_delete_scope: SyncDeleteSubscribeScope | None = None
 
 
 def configure_delete_subscribe_scope(provider: DeleteSubscribeScope) -> None:
@@ -190,3 +298,16 @@ def get_delete_subscribe_scope() -> AbstractAsyncContextManager[DeleteSubscribeC
     if _configured_delete_scope is None:
         raise RuntimeError("订阅删除事务作用域尚未配置")
     return _configured_delete_scope()
+
+
+def configure_sync_delete_subscribe_scope(provider: SyncDeleteSubscribeScope) -> None:
+    """由启动组合根登记同步消息入口使用的订阅删除事务作用域。"""
+    global _configured_sync_delete_scope
+    _configured_sync_delete_scope = provider
+
+
+def get_sync_delete_subscribe_scope() -> AbstractContextManager[SyncDeleteSubscribeCommand]:
+    """返回一次独占同步会话的订阅删除命令作用域。"""
+    if _configured_sync_delete_scope is None:
+        raise RuntimeError("同步订阅删除事务作用域尚未配置")
+    return _configured_sync_delete_scope()

app/chain/subscribe.py

@@ -48,6 +48,10 @@ from app.application.messaging.message import MessageTemplateHelper
 from app.application.mediaserver import MediaServerHelper
 from app.application.subscription.write import add_subscribe, async_add_subscribe
 from app.application.subscription.complete import get_subscription_completion_scope
+from app.application.subscription.delete import (
+    SubscribeDeletionActor,
+    get_sync_delete_subscribe_scope,
+)
 from app.application.subscription.contract import (
     build_subscribe_meta as _build_subscribe_meta,
     subscribe_media_key,
@@ -3171,9 +3175,18 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
             messenger=self,
             actions=self,
             repository=SubscribeOper(),
-            report_deleted=MoviePilotServerHelper.sub_done_async,
+            delete_subscription=self._delete_subscription,
         )
 
+    @staticmethod
+    def _delete_subscription(subscribe_id: int) -> bool:
+        """通过统一同步命令删除订阅，保留消息入口原有的全局管理权限。"""
+        with get_sync_delete_subscribe_scope() as command:
+            return command.execute(
+                subscribe_id,
+                SubscribeDeletionActor(username="", is_superuser=True),
+            )
+
     def remote_delete(self, arg_str: str, channel: NotificationChannel,
                       userid: Union[str, int] = None, source: Optional[str] = None):
         """
@@ -3189,28 +3202,18 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
                 save_history=False))
             return
         arg_strs = str(arg_str).split()
-        subscribeoper = SubscribeOper()
         for arg_str in arg_strs:
             arg_str = arg_str.strip()
             if not arg_str.isdigit():
                 continue
             subscribe_id = int(arg_str)
-            subscribe = subscribeoper.get(subscribe_id)
-            if not subscribe:
+            if not self._delete_subscription(subscribe_id):
                 self.post_message(_SchemaMessage(
                     channel=channel, source=source,
                     title=f"订阅编号 {subscribe_id} 不存在！",
                     userid=userid,
                     save_history=False))
                 return
-            # 删除订阅
-            subscribeoper.delete(subscribe_id)
-            # 统计订阅
-            MoviePilotServerHelper.sub_done_async({
-                "media_source": subscribe.media_source,
-                "media_id": subscribe.media_id,
-                "season": subscribe.season,
-            })
         # 重新发送消息
         self.remote_list(channel=channel, userid=userid, source=source)
 

app/db/oper/subscribe.py

@@ -319,22 +319,36 @@ class SubscribeOper(DbOper):
         )
 
     async def get_candidate(
-            self,
-            subscribe_id: int,
+        self,
+        subscribe_id: int,
     ) -> Optional[SubscribeDeletionCandidate]:
         """读取订阅删除用例需要的权限字段与完整事件快照。"""
         subscribe = await self.async_get(subscribe_id)
+        return self._deletion_candidate(subscribe_id, subscribe)
+
+    def get_candidate_sync(
+        self,
+        subscribe_id: int,
+    ) -> Optional[SubscribeDeletionCandidate]:
+        """同步读取订阅删除用例需要的权限字段与完整事件快照。"""
+        return self._deletion_candidate(subscribe_id, self.get(subscribe_id))
+
+    @staticmethod
+    def _deletion_candidate(
+        subscribe_id: int,
+        subscribe: Optional[Subscribe],
+    ) -> Optional[SubscribeDeletionCandidate]:
+        """把 ORM 行投影为同步和异步删除命令共用的稳定快照。"""
         if not subscribe:
             return None
         values = subscribe.__dict__
-        event_payload = {
-            column.name: values.get(column.name)
-            for column in subscribe.__table__.columns
-        }
         return SubscribeDeletionCandidate(
             subscribe_id=subscribe_id,
             username=subscribe.username,
-            event_payload=event_payload,
+            event_payload={
+                column.name: values.get(column.name)
+                for column in subscribe.__table__.columns
+            },
         )
 
     async def list_candidates_by_identity(
@@ -484,6 +498,12 @@ class SubscribeOper(DbOper):
             sqlalchemy_delete(Subscribe).where(Subscribe.id == sid)
         )
 
+    def stage_delete_sync(self, sid: int) -> None:
+        """同步登记订阅删除但不提交，由 Application UnitOfWork 控制事务边界。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("同步订阅删除需要调用方提供 Session")
+        self._db.execute(sqlalchemy_delete(Subscribe).where(Subscribe.id == sid))
+
     async def async_update(self, sid: int, payload: dict) -> Optional[Subscribe]:
         """
         异步更新订阅。

app/startup/composition/subscription.py

@@ -1,5 +1,6 @@
 """订阅事务作用域及提交后回调的组合装配。"""
 
+from collections.abc import AsyncIterator, Iterator
 from contextlib import asynccontextmanager, contextmanager
 from typing import Any
 
@@ -10,7 +11,9 @@ from app.application.subscription.complete import (
 )
 from app.application.subscription.delete import (
     DeleteSubscribeCommand,
+    SyncDeleteSubscribeCommand,
     configure_delete_subscribe_scope,
+    configure_sync_delete_subscribe_scope,
 )
 from app.application.subscription.mutation import (
     SubscriptionMutationService,
@@ -38,13 +41,18 @@ async def _publish_deleted(payload: dict[str, Any]) -> None:
     await EventManager().async_send_event(EventType.SubscribeDeleted, payload)
 
 
+def _publish_deleted_sync(payload: dict[str, Any]) -> None:
+    """为同步消息入口发布事务已提交的订阅删除事件。"""
+    EventManager().send_event(EventType.SubscribeDeleted, payload)
+
+
 def _publish_completed(payload: dict[str, Any]) -> None:
     """发布已提交的订阅完成事件。"""
     EventManager().send_event(EventType.SubscribeComplete, payload)
 
 
 @contextmanager
-def subscription_completion_scope():
+def subscription_completion_scope() -> Iterator[CompleteSubscriptionCommand]:
     """为同步完成链创建独占 Session、UoW 与 durable outbox。"""
     session = SessionFactory()
     try:
@@ -59,7 +67,7 @@ def subscription_completion_scope():
 
 
 @asynccontextmanager
-async def subscription_mutation_scope():
+async def subscription_mutation_scope() -> AsyncIterator[SubscriptionMutationService]:
     """为非 HTTP 入口创建独占订阅修改会话、UoW 与 outbox。"""
     async with async_session_scope() as session:
         yield SubscriptionMutationService(
@@ -72,7 +80,7 @@ async def subscription_mutation_scope():
 
 
 @asynccontextmanager
-async def delete_subscribe_scope():
+async def delete_subscribe_scope() -> AsyncIterator[DeleteSubscribeCommand]:
     """为非 HTTP 入口创建独占订阅删除会话、UoW 与 outbox。"""
     async with async_session_scope() as session:
         yield DeleteSubscribeCommand(
@@ -84,8 +92,25 @@ async def delete_subscribe_scope():
         )
 
 
+@contextmanager
+def sync_delete_subscribe_scope() -> Iterator[SyncDeleteSubscribeCommand]:
+    """为同步消息入口创建独占 Session、UoW 与 durable outbox。"""
+    session = SessionFactory()
+    try:
+        yield SyncDeleteSubscribeCommand(
+            repository=SubscribeOper(session),
+            unit_of_work=SqlAlchemyUnitOfWork(session),
+            publish_deleted=_publish_deleted_sync,
+            report_deleted=MoviePilotServerHelper.sub_done_durable,
+            outbox=SqlAlchemyOutboxRepository(session),
+        )
+    finally:
+        session.close()
+
+
 def configure_transactional_subscription_scopes() -> None:
     """登记 Agent 等非 HTTP 入口复用的订阅事务作用域。"""
     configure_subscription_mutation_scope(subscription_mutation_scope)
     configure_delete_subscribe_scope(delete_subscribe_scope)
+    configure_sync_delete_subscribe_scope(sync_delete_subscribe_scope)
     configure_subscription_completion_scope(subscription_completion_scope)

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～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界。
+> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口。
 
 ## 当前复核结论（2026-08-24）
 
@@ -15,7 +15,7 @@
 
 ### 长期整改阶段 0：治理门禁恢复（2026-08-23）
 
-- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异：当前为 `805` 个模块、`6502` 条内部导入边，12 组重点禁止边继续全部为 `0`，唯一非平凡 SCC 仍是隔离的 TMDB 移植包。
+- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异：当前为 `805` 个模块、`6503` 条内部导入边，12 组重点禁止边继续全部为 `0`，唯一非平凡 SCC 仍是隔离的 TMDB 移植包。
 - 启动性能探针会在隔离生命周期中真实创建并释放 TaskRegistry；normal/safe 组件数分别为 `23`/`11`，CI 只读检查使用稳定的宿主模块集合和生命周期组件顺序，不再把 Python/平台模块数量当作硬合同。
 - 官方插件快照覆盖 `plugins.v3`、`plugins.v2` 以及 V3 实际会从 `package.json` 回退加载的 31 个默认实现；`app/plugins/**` 仍只是宿主运行副本，不进入扫描。
 - SDK 快照以各模块显式 `__all__` 为公开合同，能够记录赋值别名；`typing`、`__future__` 等实现期导入不再被误冻结，既有数据库备份门面已补精确导出清单。
@@ -80,13 +80,25 @@
 - 兼容边界不变：未知第三方自定义方法继续走开放 legacy fallback；旧插件签名和结果不匹配仍只诊断、
   不拒绝加载。已登记方法的名称、kwargs、插件优先级、同步/异步入口和异常隔离 ABI 均保留。
 
+### 长期整改阶段 3：订阅删除生产路径统一（2026-08-24）
+
+- `/subscribes` 交互删除与 `/subscribe_delete` 远程命令不再直接调用 `SubscribeOper.delete` 后启动
+  `sub_done_async` 裸线程；两个同步入口统一委托 `app.application.subscription.delete` 的权限、事务、
+  事件、统计与 outbox 协议，和 API、Agent 的异步删除路径共享候选快照、授权规则及 durable intent。
+- 启动组合根为同步消息入口提供独占 Session、UoW 和 outbox；订阅行与 `subscribe.deleted`、
+  `subscribe.deleted.report` 在同一事务提交，暂存或提交失败都会回滚，成功后仍按事件、统计顺序收口。
+- 消息交互层只保留 ID 解析、名称展示和结果提示，不再拥有写库或统计实现；事务内目标已消失时按既有
+  “未找到”语义返回，避免读取与删除竞态被误报为成功。
+- 插件兼容边界保持不变：`MoviePilotServerHelper.sub_reg_async` 与 `sub_done_async` 的类方法、签名和返回值
+  继续保留给旧插件；仅宿主生产调用清零，因此没有改动插件仓、SDK/Compat 映射或事件 payload。
+
 ### 总体判断
 
 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**：
 
 - 继续采用单进程控制面是正确选择，不建议现在拆成微服务；插件、调度器、工作流、事件和数据库共享进程内状态，拆分会放大部署、事务和兼容成本。
 - `foundation/domain/runtime/adapters/application/chain/api/startup` 的职责方向基本成立；宿主架构基线、复杂度 ratchet、异步阻塞 ratchet 当前均通过。
-- 依赖图当前为 `805` 个 Python 模块、`6502` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。
+- 依赖图当前为 `805` 个 Python 模块、`6503` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。
 - 当前主要风险已经从“目录和依赖失控”转移到运行时协议、后台副作用的可靠性和遗留兼容面。换言之，下一阶段重点应是**语义收口和可验证性**，而不是继续搬文件或机械拆大文件。
 
 综合评价：架构方向可持续，生产可用性较高；可演进性仍处于中等水平。现阶段没有静态审计发现必须立即推倒重来的 P0 架构问题，但存在需要按 P1/P2 计划治理的真实债务。
@@ -104,6 +116,8 @@
    `shield` 不再让网络请求逃逸生命周期预算，仓库级并发合并、缓存键和 V1/V2/V3 返回兼容保持不变。
    请求作用域的结构化并发不进入全局登记器：传统 WebAgent SSE 的 collection 子任务改由生成器
    `finally` 取消并等待清理，断线和 ASGI 取消均不会留下请求级 task。
+   订阅删除的宿主生产者也已完成 durable 分级：消息交互和远程删除不再调用裸线程统计入口，而是
+   与业务删除原子暂存事件和统计 intent；旧类方法只作为插件 ABI 保留，不纳入宿主可靠性证明。
 2. **Model/Base 的数据库装饰器和隐式会话 ABI 已全部清零。** 查询、写事务和 `legacy_*` 装饰器均为 `0`；所有 Model `db` 参数要求显式 Session，Base CRUD 仅在调用方事务内查询或 stage。可无会话构造的入口统一留在 Oper，经组合根事务执行器运行；插件 SDK 不再导出宿主 Model。后续重点转为减少 ORM 对象跨层流转，并保持 Model 隐式事务零回退。
 
    Oper 内部的执行入口也已统一：最后一处 `AgentTaskOper` 直接 transaction runner 调用已迁入

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6502,
-  "edge_sha256": "fe24e90415afe0789875f5e6d936e9b7d3536b3bfd0383d585ea35cc4be76386",
+  "edge_count": 6503,
+  "edge_sha256": "57c326a81dbba07909871a97df1c73416850205c3d385a15afe3a07e3e44c00c",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -3324,6 +3324,7 @@
     "app.chain.subscribe -> app.application.subscription",
     "app.chain.subscribe -> app.application.subscription.complete",
     "app.chain.subscribe -> app.application.subscription.contract",
+    "app.chain.subscribe -> app.application.subscription.delete",
     "app.chain.subscribe -> app.application.subscription.query",
     "app.chain.subscribe -> app.application.subscription.write",
     "app.chain.subscribe -> app.application.torrent",

tests/fixtures/architecture/runtime-contract-baseline.json

@@ -2221,7 +2221,7 @@
           },
           {
             "caller": "app.startup.composition.subscription",
-            "count": 1
+            "count": 2
           },
           {
             "caller": "app.startup.initializers.modules",
@@ -2344,7 +2344,7 @@
         ]
       }
     },
-    "producer_count": 77
+    "producer_count": 78
   },
   "module_method_specs": {
     "anilist_credits": {

tests/test_subscribe_delete_command.py

@@ -1,11 +1,13 @@
 """订阅删除应用用例的事务、权限与副作用时序测试。"""
 
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from sqlalchemy.orm import Session
 
 from app.application.subscription.delete import (
     DeleteSubscribeCommand,
+    SyncDeleteSubscribeCommand,
     SubscribeDeletionActor,
     SubscribeDeletionCandidate,
 )
@@ -16,10 +18,11 @@ from app.db.oper.subscribe import SubscribeOper
 class _Repository:
     """记录订阅删除用例数据访问顺序的仓储替身。"""
 
-    def __init__(self, candidate, calls):
-        """保存候选订阅和共享调用序列。"""
+    def __init__(self, candidate, calls, delete_error=None):
+        """保存候选订阅、共享调用序列和可选删除异常。"""
         self.candidate = candidate
         self.calls = calls
+        self.delete_error = delete_error
 
     async def get_candidate(self, subscribe_id):
         """返回预设候选订阅。"""
@@ -29,6 +32,8 @@ class _Repository:
     async def stage_delete(self, subscribe_id):
         """记录待删除的订阅编号。"""
         self.calls.append(("delete", subscribe_id))
+        if self.delete_error:
+            raise self.delete_error
 
 
 class _UnitOfWork:
@@ -69,6 +74,62 @@ class _Outbox:
         self.calls.append(("outbox_complete", event_key))
 
 
+class _SyncRepository:
+    """记录同步订阅删除的数据访问顺序。"""
+
+    def __init__(self, candidate, calls, delete_error=None):
+        """保存候选订阅、共享调用序列和可选删除异常。"""
+        self.candidate = candidate
+        self.calls = calls
+        self.delete_error = delete_error
+
+    def get_candidate_sync(self, subscribe_id):
+        """返回预设候选订阅。"""
+        self.calls.append(("get", subscribe_id))
+        return self.candidate
+
+    def stage_delete_sync(self, subscribe_id):
+        """记录同步待删除编号并按需失败。"""
+        self.calls.append(("delete", subscribe_id))
+        if self.delete_error:
+            raise self.delete_error
+
+
+class _SyncUnitOfWork:
+    """记录同步删除命令的提交与回滚。"""
+
+    def __init__(self, calls, commit_error=None):
+        """保存共享调用序列与可选提交异常。"""
+        self.calls = calls
+        self.commit_error = commit_error
+
+    def commit(self):
+        """记录提交并按需抛出异常。"""
+        self.calls.append(("commit",))
+        if self.commit_error:
+            raise self.commit_error
+
+    def rollback(self):
+        """记录回滚。"""
+        self.calls.append(("rollback",))
+
+
+class _SyncOutbox:
+    """记录同步删除 intent 的暂存与完成顺序。"""
+
+    def __init__(self, calls):
+        """保存共享调用序列。"""
+        self.calls = calls
+
+    def stage(self, intent, _now):
+        """记录同步暂存的 intent。"""
+        self.calls.append(("outbox_stage", intent))
+
+    def complete_by_event_key(self, event_key, _completed_at):
+        """记录同步完成的 intent。"""
+        self.calls.append(("outbox_complete", event_key))
+
+
 def _candidate(username="alice"):
     """构造带完整事件身份字段的订阅删除候选。"""
     return SubscribeDeletionCandidate(
@@ -92,6 +153,7 @@ def _command(
     event_error=None,
     report_error=None,
     outbox=None,
+    delete_error=None,
 ):
     """构造可观察事件与上报失败的订阅删除用例。"""
     async def publish(payload):
@@ -108,7 +170,7 @@ def _command(
         return True
 
     return DeleteSubscribeCommand(
-        repository=_Repository(candidate, calls),
+        repository=_Repository(candidate, calls, delete_error),
         unit_of_work=_UnitOfWork(calls, commit_error),
         publish_deleted=publish,
         report_deleted=report,
@@ -138,6 +200,26 @@ def _async_report_command(candidate, calls, result=True, error=None, outbox=None
     )
 
 
+def _sync_command(candidate, calls, commit_error=None, delete_error=None, outbox=None):
+    """构造可观察事务和副作用顺序的同步订阅删除命令。"""
+    def publish(payload):
+        """记录同步删除事件。"""
+        calls.append(("event", payload["subscribe_id"], payload))
+
+    def report(payload):
+        """记录同步删除统计。"""
+        calls.append(("report", payload))
+        return True
+
+    return SyncDeleteSubscribeCommand(
+        repository=_SyncRepository(candidate, calls, delete_error),
+        unit_of_work=_SyncUnitOfWork(calls, commit_error),
+        publish_deleted=publish,
+        report_deleted=report,
+        outbox=outbox,
+    )
+
+
 @pytest.mark.asyncio
 async def test_owner_delete_commits_before_event_and_report():
     """owner 删除成功时必须先提交，再按原顺序发送事件和上报。"""
@@ -202,6 +284,26 @@ async def test_commit_failure_rolls_back_without_event_or_report():
     assert [call[0] for call in calls] == ["get", "delete", "commit", "rollback"]
 
 
+@pytest.mark.asyncio
+async def test_delete_stage_failure_rolls_back_without_effects():
+    """异步暂存失败必须显式回滚，且不得写 intent 或发送成功副作用。"""
+    calls = []
+    command = _command(
+        _candidate(),
+        calls,
+        delete_error=RuntimeError("delete failed"),
+        outbox=_Outbox(calls),
+    )
+
+    with pytest.raises(RuntimeError, match="delete failed"):
+        await command.execute(
+            7,
+            SubscribeDeletionActor(username="alice", is_superuser=False),
+        )
+
+    assert [call[0] for call in calls] == ["get", "delete", "rollback"]
+
+
 @pytest.mark.asyncio
 async def test_event_failure_happens_after_commit_and_stops_report():
     """事件失败保持原有传播语义，但事务必须已经提交且不得继续上报。"""
@@ -385,3 +487,65 @@ async def test_repository_stage_delete_does_not_commit():
 
     session.execute.assert_awaited_once()
     session.commit.assert_not_awaited()
+
+
+def test_sync_delete_uses_same_durable_effect_order():
+    """同步消息入口必须复用异步删除命令的事务、事件和统计顺序。"""
+    calls = []
+    command = _sync_command(_candidate(), calls, outbox=_SyncOutbox(calls))
+
+    assert command.execute(
+        7,
+        SubscribeDeletionActor(username="", is_superuser=True),
+    ) is True
+
+    assert [call[0] for call in calls] == [
+        "get", "delete", "outbox_stage", "outbox_stage", "commit",
+        "event", "outbox_complete", "report", "outbox_complete",
+    ]
+    assert calls[2][1].topic == "subscribe.deleted"
+    assert calls[3][1].topic == "subscribe.deleted.report"
+    assert calls[7][1] == _candidate().event_payload
+
+
+@pytest.mark.parametrize("failure", ["delete", "commit"])
+def test_sync_delete_rolls_back_transaction_failures(failure):
+    """同步暂存或提交失败时必须回滚，且不得发送删除成功副作用。"""
+    calls = []
+    error = RuntimeError(f"{failure} failed")
+    command = _sync_command(
+        _candidate(),
+        calls,
+        delete_error=error if failure == "delete" else None,
+        commit_error=error if failure == "commit" else None,
+    )
+
+    with pytest.raises(RuntimeError, match=f"{failure} failed"):
+        command.execute(
+            7,
+            SubscribeDeletionActor(username="", is_superuser=True),
+        )
+
+    assert calls[-1] == ("rollback",)
+    assert all(call[0] not in {"event", "report"} for call in calls)
+
+
+def test_sync_repository_candidate_and_delete_share_caller_session(monkeypatch):
+    """同步仓储投影和删除都使用组合根传入的同一个 Session 且不提交。"""
+    subscribe = Subscribe(
+        id=7,
+        username="alice",
+        name="测试订阅",
+        media_source="tmdb",
+        media_id="123",
+        season=2,
+    )
+    session = MagicMock(spec=Session)
+    oper = SubscribeOper(session)
+    monkeypatch.setattr(oper, "get", lambda subscribe_id: subscribe)
+
+    candidate = oper.get_candidate_sync(7)
+    oper.stage_delete_sync(7)
+
+    assert candidate is not None
+    assert candidate.event_payload["media_id"] == "123"

tests/test_subscribe_interaction_delete.py

@@ -0,0 +1,52 @@
+"""订阅消息交互删除动作的应用边界测试。"""
+
+from types import SimpleNamespace
+
+from app.application.messaging.subscribe import SubscribeInteractionHandler
+
+
+class _Repository:
+    """只提供消息展示所需读取能力的订阅仓储替身。"""
+
+    def __init__(self, subscribes):
+        """按订阅 ID 保存测试快照。"""
+        self._subscribes = subscribes
+
+    def get(self, subscribe_id):
+        """返回指定订阅快照。"""
+        return self._subscribes.get(subscribe_id)
+
+
+def test_interaction_delete_delegates_each_existing_id_to_application_action():
+    """消息层只负责解析和展示，删除必须委托统一 Application 动作。"""
+    deleted_ids = []
+    handler = SubscribeInteractionHandler(
+        messenger=SimpleNamespace(),
+        actions=SimpleNamespace(),
+        repository=_Repository({
+            7: SimpleNamespace(name="电影订阅"),
+            8: SimpleNamespace(name="剧集订阅"),
+        }),
+        delete_subscription=lambda subscribe_id: deleted_ids.append(subscribe_id) or True,
+    )
+
+    success, message = handler._delete_subscribes("7 8 9")
+
+    assert success is True
+    assert deleted_ids == [7, 8]
+    assert message == "已删除 2 个订阅：电影订阅, 剧集订阅；未找到：9"
+
+
+def test_interaction_delete_reports_transaction_race_as_missing():
+    """读取后事务内目标消失时不得误报已删除。"""
+    handler = SubscribeInteractionHandler(
+        messenger=SimpleNamespace(),
+        actions=SimpleNamespace(),
+        repository=_Repository({7: SimpleNamespace(name="电影订阅")}),
+        delete_subscription=lambda _subscribe_id: False,
+    )
+
+    success, message = handler._delete_subscribes("7")
+
+    assert success is False
+    assert message == "未找到订阅：7"