feat: make subscribe completion durable

app/application/subscription/complete.py

@@ -0,0 +1,141 @@
+"""订阅完成应用命令及其同步事务端口。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from contextlib import AbstractContextManager
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+from app.application.outbox import OutboxIntent, SyncOutboxTransaction, SyncUnitOfWork
+
+
+class SubscriptionCompletionRepository(Protocol):
+    """订阅完成命令需要的最小同步持久化端口。"""
+
+    def add_history(self, **payload: Any) -> None:
+        """在当前事务中暂存订阅历史。"""
+        ...
+
+    def delete(self, subscribe_id: int) -> None:
+        """在当前事务中暂存订阅删除。"""
+        ...
+
+
+CompletionEffect = Callable[[], None]
+CompletionReporter = Callable[[Mapping[str, Any]], object]
+
+
+class CompleteSubscriptionCommand:
+    """原子完成订阅，并按通知、事件、统计顺序执行提交后副作用。"""
+
+    def __init__(
+        self,
+        repository: SubscriptionCompletionRepository,
+        unit_of_work: SyncUnitOfWork,
+        outbox: SyncOutboxTransaction | None,
+        publish: Callable[[dict[str, Any]], None],
+    ) -> None:
+        """注入共享同步会话、事件发布端口和可选 durable outbox。"""
+        self._repository = repository
+        self._unit_of_work = unit_of_work
+        self._outbox = outbox
+        self._publish = publish
+
+    def execute(
+        self,
+        subscribe_id: int,
+        subscribe_info: Mapping[str, Any],
+        mediainfo: Mapping[str, Any],
+        notify: CompletionEffect,
+        report: CompletionReporter,
+    ) -> None:
+        """在同一事务中写历史、删订阅并暂存完成事件与统计意图。"""
+        info = dict(subscribe_info)
+        event_payload = {
+            "subscribe_id": subscribe_id,
+            "subscribe_info": info,
+            "mediainfo": dict(mediainfo),
+            "idempotency_key": completion_event_key(subscribe_id, info),
+        }
+        event_key = event_payload["idempotency_key"]
+        report_key = completion_report_key(subscribe_id, info)
+        report_payload = {"subscribe_info": _completion_report_payload(info, report_key)}
+        try:
+            self._repository.add_history(**info)
+            self._repository.delete(subscribe_id)
+            if self._outbox:
+                now = datetime.now(timezone.utc)
+                self._outbox.stage(
+                    OutboxIntent(
+                        event_key=event_key,
+                        topic="subscribe.complete",
+                        payload=event_payload,
+                    ),
+                    now,
+                )
+                self._outbox.stage(
+                    OutboxIntent(
+                        event_key=report_key,
+                        topic="subscribe.complete.report",
+                        payload=report_payload,
+                    ),
+                    now,
+                )
+            self._unit_of_work.commit()
+        except Exception:
+            self._unit_of_work.rollback()
+            raise
+
+        notify()
+        self._publish(event_payload)
+        if self._outbox:
+            self._outbox.complete_by_event_key(event_key, datetime.now(timezone.utc))
+        if report(report_payload["subscribe_info"]) is False:
+            raise RuntimeError("订阅完成统计上报未确认")
+        if self._outbox:
+            self._outbox.complete_by_event_key(report_key, datetime.now(timezone.utc))
+
+
+def completion_event_key(subscribe_id: int, subscribe_info: Mapping[str, Any]) -> str:
+    """构造跨重试稳定的订阅完成事件幂等键。"""
+    return (
+        f"subscribe.complete:{subscribe_id}:"
+        f"{subscribe_info.get('media_source') or 'unknown'}:"
+        f"{subscribe_info.get('media_id') or 'unknown'}:v1"
+    )
+
+
+def completion_report_key(subscribe_id: int, subscribe_info: Mapping[str, Any]) -> str:
+    """构造可独立重试的订阅完成统计幂等键。"""
+    return f"{completion_event_key(subscribe_id, subscribe_info)}:report"
+
+
+def _completion_report_payload(
+    subscribe_info: Mapping[str, Any],
+    report_key: str,
+) -> dict[str, Any]:
+    """保留旧统计接口字段，同时为恢复 handler 固化幂等键。"""
+    return {
+        "media_source": subscribe_info.get("media_source"),
+        "media_id": subscribe_info.get("media_id"),
+        "season": subscribe_info.get("season"),
+        "idempotency_key": report_key,
+    }
+
+
+CompletionScope = Callable[[], AbstractContextManager[CompleteSubscriptionCommand]]
+_configured_completion_scope: CompletionScope | None = None
+
+
+def configure_subscription_completion_scope(provider: CompletionScope) -> None:
+    """由启动组合根登记订阅完成独占事务作用域。"""
+    global _configured_completion_scope
+    _configured_completion_scope = provider
+
+
+def get_subscription_completion_scope() -> AbstractContextManager[CompleteSubscriptionCommand]:
+    """返回一次独占同步订阅完成事务作用域。"""
+    if _configured_completion_scope is None:
+        raise RuntimeError("订阅完成事务作用域尚未配置")
+    return _configured_completion_scope()

app/chain/subscribe.py

@@ -46,6 +46,7 @@ from app.application.configuration import (
 from app.application.messaging.subscribe import SubscribeInteractionHandler
 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.contract import (
     build_subscribe_meta as _build_subscribe_meta,
     subscribe_media_key,
@@ -3096,44 +3097,39 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
         # 完成订阅
         msgstr = "订阅" if not subscribe.best_version else "洗版"
         logger.info(f'{mediainfo.title_year} 完成{msgstr}')
-        # 新增订阅历史
-        subscribeoper = SubscribeOper()
-        subscribeoper.add_history(**subscribe.to_dict())
-        # 删除订阅
-        subscribeoper.delete(subscribe.id)
-        # 发送通知
+
+        # 完成命令在同一 Session/UoW 中写历史、删除订阅并暂存可恢复副作用。
         if mediainfo.type == MediaType.TV:
             link = self.runtime_config.television_subscribe_url
         elif mediainfo.type == MediaType.MUSIC:
             link = self.runtime_config.music_subscribe_url
         else:
             link = self.runtime_config.movie_subscribe_url
-        # 完成订阅按规则发送消息
-        self.post_message(
-            _SchemaMessage(
-                mtype=MessageType.Subscribe,
-                ctype=ContentType.SubscribeComplete,
-                image=mediainfo.get_message_image(),
-                link=link,
-                username=subscribe.username
-            ),
-            meta=meta,
-            mediainfo=mediainfo,
-            msgstr=msgstr,
-            username=subscribe.username
-        )
-        # 发送事件
-        eventmanager.send_event(EventType.SubscribeComplete, {
-            "subscribe_id": subscribe.id,
-            "subscribe_info": subscribe.to_dict(),
-            "mediainfo": mediainfo.to_dict(),
-        })
-        # 统计订阅
-        MoviePilotServerHelper.sub_done_async({
-            "media_source": subscribe.media_source,
-            "media_id": subscribe.media_id,
-            "season": subscribe.season,
-        })
+
+        def notify() -> None:
+            """提交成功后发送完成通知，保持历史消息 ABI。"""
+            self.post_message(
+                _SchemaMessage(
+                    mtype=MessageType.Subscribe,
+                    ctype=ContentType.SubscribeComplete,
+                    image=mediainfo.get_message_image(),
+                    link=link,
+                    username=subscribe.username,
+                ),
+                meta=meta,
+                mediainfo=mediainfo,
+                msgstr=msgstr,
+                username=subscribe.username,
+            )
+
+        with get_subscription_completion_scope() as command:
+            command.execute(
+                subscribe_id=subscribe.id,
+                subscribe_info=subscribe.to_dict(),
+                mediainfo=mediainfo.to_dict(),
+                notify=notify,
+                report=MoviePilotServerHelper.sub_done_durable,
+            )
 
     def _interaction_handler(self) -> "SubscribeInteractionHandler":
         """构造 /subscribes 交互处理器，业务动作由本链提供。"""

app/startup/modules_initializer.py

@@ -249,6 +249,13 @@ def _build_outbox_dispatcher() -> OutboxDispatcher:
         ):
             raise RuntimeError("订阅新增统计上报未确认")
 
+    def dispatch_subscribe_complete_report(message) -> None:
+        """重放订阅完成统计；未确认时抛错以进入有限重试。"""
+        if not MoviePilotServerHelper.sub_done_durable(
+            message.payload.get("subscribe_info") or {}
+        ):
+            raise RuntimeError("订阅完成统计上报未确认")
+
     session = SessionFactory()
     return OutboxDispatcher(
         repository=SqlAlchemyOutboxRepository(session),
@@ -267,6 +274,11 @@ def _build_outbox_dispatcher() -> OutboxDispatcher:
                 message.payload,
             ),
             "subscribe.deleted.report": dispatch_subscribe_deleted_report,
+            "subscribe.complete": lambda message: EventManager().send_event(
+                EventType.SubscribeComplete,
+                message.payload,
+            ),
+            "subscribe.complete.report": dispatch_subscribe_complete_report,
             "download.added": lambda message: EventManager().send_event(
                 EventType.DownloadAdded,
                 restore_download_added(message.payload),

app/startup/subscription.py

@@ -1,7 +1,7 @@
 """订阅写入事务适配器的启动装配。"""
 
 from collections.abc import Callable
-from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager
 from datetime import datetime, timezone
 from typing import Any
 
@@ -20,6 +20,10 @@ from app.application.subscription.delete import (
     DeleteSubscribeCommand,
     configure_delete_subscribe_scope,
 )
+from app.application.subscription.complete import (
+    CompleteSubscriptionCommand,
+    configure_subscription_completion_scope,
+)
 from app.application.subscription.mutation import (
     SubscriptionMutationService,
     configure_subscription_mutation_scope,
@@ -28,6 +32,7 @@ from app.adapters.external.server import MoviePilotServerHelper
 from app.db.oper.subscribe import SubscribeOper
 from app.db.oper.subscribehistory import SubscribeHistoryOper
 from app.db.session import async_session_scope
+from app.db.session import SessionFactory
 from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
 from app.startup.outbox import (
     SqlAlchemyAsyncOutboxStager,
@@ -133,6 +138,26 @@ async def _publish_deleted(payload: dict[str, Any]) -> None:
     await EventManager().async_send_event(EventType.SubscribeDeleted, payload)
 
 
+def _publish_completed(payload: dict[str, Any]) -> None:
+    """发布已提交的订阅完成事件。"""
+    EventManager().send_event(EventType.SubscribeComplete, payload)
+
+
+@contextmanager
+def subscription_completion_scope():
+    """为同步完成链创建独占 Session、UoW 与 durable outbox。"""
+    session = SessionFactory()
+    try:
+        yield CompleteSubscriptionCommand(
+            repository=SubscribeOper(session),
+            unit_of_work=SqlAlchemyUnitOfWork(session),
+            outbox=SqlAlchemyOutboxRepository(session),
+            publish=_publish_completed,
+        )
+    finally:
+        session.close()
+
+
 @asynccontextmanager
 async def subscription_mutation_scope():
     """为非 HTTP 入口创建独占订阅修改会话、UoW 与 outbox。"""
@@ -163,3 +188,4 @@ def configure_transactional_subscription_scopes() -> None:
     """登记 Agent 等非 HTTP 入口复用的订阅事务作用域。"""
     configure_subscription_mutation_scope(subscription_mutation_scope)
     configure_delete_subscribe_scope(delete_subscribe_scope)
+    configure_subscription_completion_scope(subscription_completion_scope)

docs/adr/0007-background-action-reliability.md

@@ -19,9 +19,11 @@ MoviePilot 保持模块化单体，不把所有后台动作迁到分布式队列
 `durable-required` 是目标语义，不代表当前实现已经 durable。ARCH-251 前，Event Registry 中标记该值的
 事件仍应在风险报告中说明崩溃窗口。
 
-截至 2026-08-22，宿主正式装配的 `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`、
-`DownloadAdded`、`TransferComplete`、`TransferFailed` 广播已由业务事务内的 outbox intent 提供
-at-least-once 恢复；payload 保持插件 dict/对象 ABI，并增加可选幂等键。下载和整理的 outbox 只保存
+截至 2026-08-23，宿主正式装配的 `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`、
+`SubscribeComplete`、`DownloadAdded`、`TransferComplete`、`TransferFailed` 广播已由业务事务内的 outbox
+intent 提供 at-least-once 恢复；订阅完成的历史新增、订阅删除、完成事件和完成统计 intent 同事务提交，
+提交后通知/事件/统计仍按原顺序执行，事件与统计失败保持独立 pending。payload 保持插件 dict/对象 ABI，
+并增加可选幂等键。下载和整理的 outbox 只保存
 可 JSON 序列化的快照，重放时恢复旧对象字段。这不覆盖第三方插件自行发送的裸事件，也不代表订阅通知
 和外部统计上报已经全部 durable。
 

docs/refactor/backend-architecture-next-stage.md

@@ -72,7 +72,7 @@ MoviePilot V3 当前不是“目录混乱、必须推倒重来”的状态。第
 | 专用 EventData model | 53 | Event Contract Registry 已为全部事件登记 typed payload/fallback 原因 |
 | 直接读取 `settings` 的文件 | 127 | 仍按模块族迁移，动态协议和安全端口暂保留 |
 | `SystemConfigOper()` | 1 个 | 仅组合根创建 `SystemConfigService` 时保留 |
-| Model 上的 DB 查询装饰器 | 121 | `db_update`/`async_db_update` 为 0；查询 ABI 继续按 canonical 用例迁移 |
+| Model 上的 DB 查询装饰器 | 119 | `db_update`/`async_db_update` 为 0；查询 ABI 继续按 canonical 用例迁移 |
 | 路由端点 | 335 | 11 个已装饰端点超过 80 行，最大 400 行 |
 | Chain 方法超过 150 行 | 18 | 最大 `TransferChain.do_transfer()` 885 行 |
 | Application 方法超过 150 行 | 8 | 最大 296 行 |
@@ -430,8 +430,10 @@ flowchart TB
   并继续保留既有 Model/旧 SDK 查询兼容；本次 AgentTask 切片将查询装饰器减少到 121 个。
   2026-08-23 已完成 AgentTask 查询切片：`AgentTaskOper.get/list` 直接在调用方 Session 中执行查询，
   `AgentTask.get_for_user/list_for_user` 保留原签名和返回语义供旧调用方使用，但不再持有查询装饰器；
-  无 Session 的旧 Oper 入口继续由组合根兼容事务执行器承接。查询装饰器低水位由 123 降至 121，
-  归属过滤、启用状态过滤和创建时间/主键稳定排序由 canonical Oper 测试覆盖。
+  无 Session 的旧 Oper 入口继续由组合根兼容事务执行器承接。随后 PassKey 的宿主同步查询迁移到
+  `PassKeyOper`，其按用户/凭证的启用状态过滤由显式 Session 测试覆盖；异步 Model 查询保留旧 ABI。
+  查询装饰器低水位由 123 降至 119，归属过滤、启用状态过滤和创建时间/主键稳定排序由 canonical
+  Oper 测试覆盖。
 
 #### ARCH-222：按风险迁移其余写用例
 
@@ -794,8 +796,13 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas
   对插件仍投递原有 dict 字段，只新增可选 `idempotency_key`，不把 Pydantic 实例传给插件。
 - 保证边界只覆盖主仓可追踪的宿主生产者。运行时安装在 `app/plugins/**` 的第三方插件未被主仓改写；
   插件若自行直接发送同名事件，该发送仍由插件负责，无法与插件自己的数据库写入自动组成原子事务。
-- 订阅外部统计上报仍是 post-commit 副作用，不在事件 intent 的重放 handler 中；因此当前可以宣称三种
-  订阅事件具备宿主级 at-least-once 恢复，但不能宣称订阅通知和所有外部上报均已 durable。
+- `SubscribeChain` 完成流程现由 `app/application/subscription/complete.py` 统一收口：订阅历史新增、
+  原订阅删除、`subscribe.complete` 事件 intent 与 `subscribe.complete.report` 统计 intent 在同一同步
+  Session/UoW 中提交。提交后仍按通知、事件、统计的历史顺序执行；事件和统计分别按幂等键收口，任一步
+  失败都会留下独立 pending intent，由 outbox dispatcher 有限重试并最终进入 dead-letter。完成事件仍向插件
+  投递原有 `subscribe_id`、`subscribe_info`、`mediainfo` 字段，仅增加可选 `idempotency_key`。
+- 普通订阅新增/修改/删除路径的用户通知与第三方插件自行发送的事件仍不自动纳入宿主事务；本切片只覆盖
+  主仓可追踪的 `SubscribeChain` 完成生产者。
 - `DownloadAdded`、`TransferComplete`、`TransferFailed` 也已逐项接入，而不是复用一个不分业务语义的
   “万能消息总线”。下载历史、下载文件清单或整理历史与各自 intent 在独占同步 Session/UoW 中原子提交；
   即时广播失败时 intent 保持 pending，三种恢复 handler 均继续使用有限重试与 dead-letter 策略。
@@ -815,7 +822,7 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas
   事务低水位从 174 降到 168，Oper 仍不创建 Session、也不直接 commit/rollback。
 - 剩余 45 个同步/异步 Model 写装饰器已全部迁移：AgentTask、PassKey、User、消息、历史清理、
   站点快照、媒体服务器、插件数据、TransferPending 等写入由调用方 Session 和 UoW 收口；无 Session
-  的旧 Oper ABI 委托 Startup 注入的短事务执行器。当前 Model 装饰器仅剩 123 个查询装饰器，
+  的旧 Oper ABI 委托 Startup 注入的短事务执行器。当前 Model 装饰器仅剩 119 个查询装饰器，
   `db_update` 与 `async_db_update` 均为 0，Oper 自建 Session/直接提交仍为 0。
 - 数据清理按批次显式提交 UoW，单表失败先回滚会话再继续汇总后续表；不再依赖删除 Model 的隐式提交。
 - 收尾批次进一步移除宿主 Oper 对 `Base.create/update/delete/truncate` 八个兼容包装器的调用：显式
@@ -1006,6 +1013,7 @@ MFA/Passkey 专项测试与架构门禁通过，密钥类配置仍保留在安
 
 2026-08-23 将工作流动作 `FetchRssAction`、`ScanFileAction` 和 `AddSubscribeAction` 接入 `ChainRuntimeConfig` 快照，分别移除代理、媒体后缀和超级用户的全局 `settings` 读取；保留动作公开入口与工作流上下文行为，新增快照注入测试覆盖。配置债务由 130 个文件降至 127 个文件，宿主依赖与配置基线已更新。
 2026-08-23 将工作流动作 `FetchMediasAction` 和 `SendMessageAction` 接入 `ChainRuntimeConfig` 快照，分别移除内部 API 端口/令牌及工作流链接的全局 `settings` 读取；保留动作公开入口与消息载荷行为，新增快照注入测试覆盖。配置债务由 127 个文件降至 125 个文件，宿主依赖与配置基线已更新。
+2026-08-23 将 API 路由前缀作为组合根参数传入 `init_routers`，移除路由初始化模块对全局 `settings` 的直接读取；默认参数保留旧调用兼容性，并补充自定义前缀测试。配置债务由 125 个文件降至 124 个文件。
 
 **收口记录（2026-08-22）**：`reidentify_cache`、`nettest`、`scrape`、OpenAI `chat_completions/responses`、`get_logging` 和 Web Agent SSE 均改为稳定公开入口委托私有编排实现；四个消息交互 Handler 的公开方法也保留 ABI 并委托私有状态机。复杂度基线已清零，API/Application/Chain 入口预算、异步阻塞 ratchet 均通过；复杂度及兼容专项合计 252 项测试通过。
 随后将 `TransferChain.do_transfer` 的公开入口收口为稳定兼容 Facade，先提取媒体身份规范化阶段，保留显式
@@ -1196,7 +1204,7 @@ rollback:
 | 基线写入行为 | 默认命令可能覆盖 fixture | 所有默认/check 命令保证工作树不变；write 必须显式 scope |
 | 全功能 worker | 配置允许 >1，控制面会复制 | 启动期明确拒绝 >1；文档与配置一致 |
 | 健康接口 | 认证 `/system/ping` 为主 | 分离公开 live 与受限/安全 ready；失败原因可诊断 |
-| Model 事务装饰器 | 当前 121 个且全部只读；写装饰器 0 | 查询债务只降不增；写事务不回退到 Model/Base 隐式提交 |
+| Model 事务装饰器 | 当前 119 个且全部只读；写装饰器 0 | 查询债务只降不增；写事务不回退到 Model/Base 隐式提交 |
 | 新写用例事务 | 宿主写 Oper 已脱离 Base 隐式提交 | 100% 由入口/Application 边界拥有 Session/UoW |
 | 高频 Module 契约 | 212 个宿主能力显式登记 | 新观察到的宿主方法必须同步登记完整契约 |
 | Event payload | 53 类型全部登记 typed payload 与可靠性 | 新事件必须同步登记，不回退裸 dict |

tests/test_subscription_completion_command.py

@@ -0,0 +1,141 @@
+"""订阅完成命令的原子写入、时序与 durable intent 测试。"""
+
+from datetime import datetime
+
+import pytest
+
+from app.application.subscription.complete import CompleteSubscriptionCommand
+
+
+class _Repository:
+    """记录历史暂存和订阅删除顺序。"""
+
+    def __init__(self, calls: list[tuple]) -> None:
+        """保存共享调用序列。"""
+        self.calls = calls
+
+    def add_history(self, **payload) -> None:
+        """记录历史快照。"""
+        self.calls.append(("history", payload))
+
+    def delete(self, subscribe_id: int) -> None:
+        """记录待删除订阅。"""
+        self.calls.append(("delete", subscribe_id))
+
+
+class _UnitOfWork:
+    """记录提交和回滚。"""
+
+    def __init__(self, calls: list[tuple], error: Exception | None = None) -> None:
+        """保存调用序列与可选提交异常。"""
+        self.calls = calls
+        self.error = error
+
+    def commit(self) -> None:
+        """记录提交并按需失败。"""
+        self.calls.append(("commit",))
+        if self.error:
+            raise self.error
+
+    def rollback(self) -> None:
+        """记录回滚。"""
+        self.calls.append(("rollback",))
+
+
+class _Outbox:
+    """记录 intent 暂存与即时收口。"""
+
+    def __init__(self, calls: list[tuple]) -> None:
+        """保存共享调用序列。"""
+        self.calls = calls
+
+    def stage(self, intent, _now: datetime) -> None:
+        """记录 durable intent。"""
+        self.calls.append(("stage", intent))
+
+    def complete_by_event_key(self, event_key: str, _now: datetime) -> None:
+        """记录成功副作用对应的 intent 收口。"""
+        self.calls.append(("complete", event_key))
+
+
+def _command(calls: list[tuple], *, publish_error=None, report_result=True, notify_error=None):
+    """构造可注入失败的完成命令。"""
+    def notify() -> None:
+        """记录通知。"""
+        calls.append(("notify",))
+        if notify_error:
+            raise notify_error
+
+    def publish(payload) -> None:
+        """记录完成事件。"""
+        calls.append(("event", payload))
+        if publish_error:
+            raise publish_error
+
+    def report(payload) -> bool:
+        """记录完成统计。"""
+        calls.append(("report", payload))
+        return report_result
+
+    return CompleteSubscriptionCommand(
+        repository=_Repository(calls),
+        unit_of_work=_UnitOfWork(calls),
+        outbox=_Outbox(calls),
+        publish=publish,
+    ), notify, report
+
+
+@pytest.mark.parametrize("failure", ["event", "report", "notify"])
+def test_completion_stages_business_and_independent_intents_before_commit(failure):
+    """完成事务先提交业务和两个 intent，提交后按通知、事件、统计顺序执行。"""
+    calls = []
+    command, notify, report = _command(
+        calls,
+        publish_error=RuntimeError("event failed") if failure == "event" else None,
+        report_result=False if failure == "report" else True,
+        notify_error=RuntimeError("notify failed") if failure == "notify" else None,
+    )
+
+    with pytest.raises(RuntimeError):
+        command.execute(
+            7,
+            {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2},
+            {"title": "Test"},
+            notify=notify,
+            report=report,
+        )
+
+    assert [call[0] for call in calls[:5]] == [
+        "history", "delete", "stage", "stage", "commit",
+    ]
+    assert calls[2][1].topic == "subscribe.complete"
+    assert calls[3][1].topic == "subscribe.complete.report"
+    if failure == "notify":
+        assert [call[0] for call in calls[5:]] == ["notify"]
+    elif failure == "event":
+        assert [call[0] for call in calls[5:]] == ["notify", "event"]
+    else:
+        assert [call[0] for call in calls[5:]] == [
+            "notify", "event", "complete", "report",
+        ]
+
+
+def test_completion_success_closes_event_then_report_intent():
+    """成功完成按兼容顺序通知、事件、统计，并分别收口两个 intent。"""
+    calls = []
+    command, notify, report = _command(calls)
+
+    command.execute(
+        7,
+        {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2},
+        {"title": "Test"},
+        notify=notify,
+        report=report,
+    )
+
+    assert [call[0] for call in calls] == [
+        "history", "delete", "stage", "stage", "commit",
+        "notify", "event", "complete", "report", "complete",
+    ]
+    assert calls[6][1]["idempotency_key"] == calls[2][1].event_key
+    assert calls[8][1]["idempotency_key"] == calls[3][1].event_key