refactor: make download and transfer events durable

app/application/chain/context.py

@@ -7,6 +7,7 @@ from dataclasses import dataclass
 from typing import Any, Optional
 
 from app.application.chain.data import ChainDataPorts
+from app.application.chain.durable_events import ChainDurableEventWriter
 
 
 MessageQueueFactory = Callable[[Callable[..., Any]], Any]
@@ -28,6 +29,7 @@ class ChainRuntimeContext:
     message_queue_factory: MessageQueueFactory
     module_dispatcher_factory: ModuleDispatcherFactory
     data_ports: Optional[ChainDataPorts] = None
+    durable_event_writer: Optional[ChainDurableEventWriter] = None
 
 
 def _unconfigured_chain_runtime_context() -> ChainRuntimeContext:

app/application/chain/durable_events.py

@@ -0,0 +1,230 @@
+"""Chain durable 事件的事务写端口与可重放 payload 转换。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass, fields
+from datetime import date, datetime
+from enum import Enum
+from pathlib import Path
+from typing import Any, Protocol, cast
+
+from app.application.history import TransferHistoryRecord, TransferHistoryWriter
+from app.domain.context import Context, MediaInfo, MusicInfo, TorrentInfo
+from app.domain.meta.metabase import MetaBase
+from app.domain.meta.metamusic import MetaMusic
+from app.domain.metainfo import MetaInfo
+from app.schemas.file import FileItem
+from app.schemas.transfer import TransferInfo
+from app.schemas.types import MediaType
+
+
+class ChainDurableEventWriter(Protocol):
+    """下载与整理 Chain 原子写业务记录和 outbox 的宿主端口。"""
+
+    def download_added(
+        self,
+        *,
+        history_payload: dict[str, Any],
+        file_payloads: list[dict[str, Any]],
+        event_payload: dict[str, Any],
+        after_commit: Callable[[], None],
+        publish: Callable[[dict[str, Any]], None],
+    ) -> None:
+        """提交下载历史与 DownloadAdded intent，再执行原有提交后编排。"""
+
+    def transfer_result(
+        self,
+        *,
+        topic: str,
+        stage_history: Callable[[TransferHistoryWriter], TransferHistoryRecord | None],
+        event_payload: dict[str, Any],
+        publish: Callable[[dict[str, Any]], None],
+    ) -> TransferHistoryRecord | None:
+        """提交整理历史与结果 intent，并在提交后广播兼容事件。"""
+
+
+@dataclass(frozen=True, slots=True)
+class TransferHistoryRef:
+    """事务关闭后仍可安全读取的最小整理历史投影。"""
+
+    id: int
+    status: bool
+    src: str | None
+    src_storage: str | None
+    src_fileitem: dict[str, Any] | None
+
+
+def download_added_event_key(payload: dict[str, Any]) -> str:
+    """由下载器与任务 hash 构造重试期间稳定的 DownloadAdded 幂等键。"""
+    return (
+        f"download.added:{payload.get('downloader') or 'unknown'}:"
+        f"{payload.get('hash') or 'unknown'}:v1"
+    )
+
+
+def transfer_result_event_key(topic: str, history_id: int) -> str:
+    """由结果 topic 与整理历史 ID 构造稳定幂等键。"""
+    return f"{topic}:{history_id}:v1"
+
+
+def snapshot_download_added(payload: dict[str, Any]) -> dict[str, Any]:
+    """把插件运行时 Context 转为 outbox 可 JSON 序列化的稳定快照。"""
+    context = payload.get("context")
+    return cast(dict[str, Any], _json_snapshot({
+        "hash": payload.get("hash"),
+        "context": context.to_dict() if isinstance(context, Context) else context,
+        "username": payload.get("username"),
+        "downloader": payload.get("downloader"),
+        "episodes": list(payload.get("episodes") or []),
+        "source": payload.get("source"),
+        "idempotency_key": payload.get("idempotency_key"),
+    }))
+
+
+def restore_download_added(payload: dict[str, Any]) -> dict[str, Any]:
+    """从 outbox 快照恢复插件既有 DownloadAdded 运行时对象形状。"""
+    restored = dict(payload)
+    context = payload.get("context")
+    if isinstance(context, dict):
+        restored["context"] = _restore_context(context)
+    return restored
+
+
+def snapshot_transfer_result(payload: dict[str, Any]) -> dict[str, Any]:
+    """把整理事件中的领域对象转换为可恢复 JSON 快照。"""
+    return cast(dict[str, Any], _json_snapshot({
+        "fileitem": _model_snapshot(payload.get("fileitem")),
+        "meta": _object_snapshot(payload.get("meta")),
+        "mediainfo": _object_snapshot(payload.get("mediainfo")),
+        "transferinfo": _model_snapshot(payload.get("transferinfo")),
+        "downloader": payload.get("downloader"),
+        "download_hash": payload.get("download_hash"),
+        "transfer_history_id": payload.get("transfer_history_id"),
+        "idempotency_key": payload.get("idempotency_key"),
+    }))
+
+
+def restore_transfer_result(payload: dict[str, Any]) -> dict[str, Any]:
+    """从 outbox 快照恢复 TransferComplete/Failed 的旧对象 payload。"""
+    restored = dict(payload)
+    fileitem = payload.get("fileitem")
+    meta = payload.get("meta")
+    mediainfo = payload.get("mediainfo")
+    transferinfo = payload.get("transferinfo")
+    restored["fileitem"] = (
+        FileItem.model_validate(fileitem) if isinstance(fileitem, dict) else fileitem
+    )
+    restored["meta"] = _restore_meta(meta) if isinstance(meta, dict) else meta
+    restored["mediainfo"] = (
+        _restore_media(mediainfo) if isinstance(mediainfo, dict) else mediainfo
+    )
+    restored["transferinfo"] = (
+        TransferInfo.model_validate(transferinfo)
+        if isinstance(transferinfo, dict)
+        else transferinfo
+    )
+    return restored
+
+
+def _model_snapshot(value: Any) -> Any:
+    """序列化 Pydantic 风格对象，空值和已有 JSON 值原样返回。"""
+    if hasattr(value, "model_dump"):
+        return value.model_dump(mode="json")
+    return value
+
+
+def _object_snapshot(value: Any) -> Any:
+    """序列化领域对象，避免把不可持久化实例写入 JSON 列。"""
+    if hasattr(value, "to_dict"):
+        return value.to_dict()
+    return _model_snapshot(value)
+
+
+def _json_snapshot(value: Any) -> Any:
+    """递归归一化快照，确保 SQLAlchemy JSON 不接收运行时专用对象。"""
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    if isinstance(value, Enum):
+        return _json_snapshot(value.value)
+    if isinstance(value, (Path, date, datetime)):
+        return value.isoformat() if hasattr(value, "isoformat") else str(value)
+    if isinstance(value, dict):
+        return {
+            str(key): _json_snapshot(item)
+            for key, item in value.items()
+        }
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return [_json_snapshot(item) for item in value]
+    if hasattr(value, "model_dump"):
+        return _json_snapshot(value.model_dump(mode="json"))
+    if hasattr(value, "to_dict"):
+        return _json_snapshot(value.to_dict())
+    return str(value)
+
+
+def _restore_context(payload: dict[str, Any]) -> Context:
+    """恢复 DownloadAdded 插件依赖的 Context 聚合对象。"""
+    meta_payload = payload.get("meta_info")
+    media_payload = payload.get("media_info")
+    torrent_payload = payload.get("torrent_info")
+    allowed_episodes = payload.get("allowed_episodes")
+    return Context(
+        meta_info=(
+            _restore_meta(meta_payload) if isinstance(meta_payload, dict) else None
+        ),
+        media_info=(
+            _restore_media(media_payload) if isinstance(media_payload, dict) else None
+        ),
+        torrent_info=(
+            _restore_torrent(torrent_payload)
+            if isinstance(torrent_payload, dict)
+            else None
+        ),
+        media_recognize_fail_count=int(
+            payload.get("media_recognize_fail_count") or 0
+        ),
+        resource_source=str(payload.get("resource_source") or "unknown"),
+        match_source=str(payload.get("match_source") or "unknown"),
+        candidate_recognized=bool(payload.get("candidate_recognized")),
+        media_info_is_target=bool(payload.get("media_info_is_target")),
+        allowed_episodes=(
+            set(allowed_episodes) if allowed_episodes is not None else None
+        ),
+        confirmed_full_coverage=bool(payload.get("confirmed_full_coverage")),
+    )
+
+
+def _restore_meta(payload: dict[str, Any]) -> MetaBase:
+    """恢复影视或音乐文件名解析对象，并保留快照中的解析字段。"""
+    if payload.get("type") in {MediaType.MUSIC, MediaType.MUSIC.value, "music"}:
+        return MetaMusic.from_dict(payload)
+    title = str(
+        payload.get("org_string")
+        or payload.get("title")
+        or payload.get("name")
+        or ""
+    )
+    meta = MetaInfo(title)
+    for key, value in payload.items():
+        if key in {"season_episode", "edition", "name", "episode_list"}:
+            continue
+        if key == "type" and value:
+            value = MediaType(value)
+        setattr(meta, key, value)
+    return meta
+
+
+def _restore_media(payload: dict[str, Any]) -> MediaInfo | MusicInfo:
+    """恢复影视或音乐媒体对象，不触发任何远端识别。"""
+    if payload.get("type") in {MediaType.MUSIC, MediaType.MUSIC.value, "music"}:
+        return MusicInfo.from_dict(payload)
+    media = MediaInfo()
+    media.from_dict(payload)
+    return media
+
+
+def _restore_torrent(payload: dict[str, Any]) -> TorrentInfo:
+    """按 dataclass 构造字段恢复种子对象，忽略快照中的计算属性。"""
+    allowed = {item.name for item in fields(TorrentInfo) if item.init}
+    return TorrentInfo(**{key: value for key, value in payload.items() if key in allowed})

app/application/outbox.py

@@ -5,7 +5,10 @@ from __future__ import annotations
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import Any, Protocol
+from typing import Any, Protocol, TypeVar
+
+
+T = TypeVar("T")
 
 
 @dataclass(frozen=True, slots=True)
@@ -67,6 +70,70 @@ class AsyncOutboxTransaction(Protocol):
         """即时投递成功后按稳定幂等键标记 intent 完成。"""
 
 
+class SyncUnitOfWork(Protocol):
+    """同步 durable 业务切片的最小事务端口。"""
+
+    def commit(self) -> None:
+        """提交业务写入与 outbox intent。"""
+
+    def rollback(self) -> None:
+        """回滚业务写入与 outbox intent。"""
+
+
+class SyncOutboxTransaction(Protocol):
+    """同步业务事务暂存并收口 durable intent 的最小端口。"""
+
+    def stage(self, intent: OutboxIntent, now: datetime) -> None:
+        """把 intent 加入调用方事务，但不自行提交。"""
+
+    def complete_by_event_key(
+        self,
+        event_key: str,
+        completed_at: datetime,
+    ) -> None:
+        """即时投递成功后按幂等键标记 intent 完成。"""
+
+
+class DurableEventCommand:
+    """把一次同步业务写入与可恢复事件 intent 原子提交。"""
+
+    def __init__(
+        self,
+        unit_of_work: SyncUnitOfWork,
+        outbox: SyncOutboxTransaction,
+    ) -> None:
+        """注入共享同一 Session 的事务与 outbox 端口。"""
+        self._unit_of_work = unit_of_work
+        self._outbox = outbox
+
+    def execute(
+        self,
+        *,
+        intent: OutboxIntent | Callable[[T], OutboxIntent],
+        stage_business: Callable[[], T],
+        publish: Callable[[], None],
+        after_commit: Callable[[], None] | None = None,
+    ) -> T:
+        """先原子提交业务与 intent，再保持原顺序执行提交后动作和即时广播。"""
+        try:
+            result = stage_business()
+            resolved_intent = intent(result) if callable(intent) else intent
+            self._outbox.stage(resolved_intent, datetime.now(timezone.utc))
+            self._unit_of_work.commit()
+        except Exception:
+            self._unit_of_work.rollback()
+            raise
+
+        if after_commit:
+            after_commit()
+        publish()
+        self._outbox.complete_by_event_key(
+            resolved_intent.event_key,
+            datetime.now(timezone.utc),
+        )
+        return result
+
+
 class OutboxDispatcher:
     """认领并派发 outbox，按 event key 依赖 handler 幂等。"""
 

app/chain/__init__.py

@@ -53,6 +53,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
         self.filecache = context.file_cache
         self.async_filecache = context.async_file_cache
         self.data_ports = context.data_ports or get_chain_data_ports()
+        self.durable_event_writer = context.durable_event_writer
         self._module_dispatcher = context.module_dispatcher_factory(
             module_catalog=self.modulemanager,
             plugin_catalog=self.pluginmanager,

app/chain/download.py

@@ -105,6 +105,45 @@ class DownloadChain(ChainBase):
         }
         return note
 
+    def _after_download_history_commit(
+        self,
+        *,
+        context: Context,
+        media: MediaInfo | MusicInfo,
+        meta: MetaBase,
+        torrent: TorrentInfo,
+        channel: NotificationChannel | None,
+        source: str | None,
+        userid: str | None,
+        username: str | None,
+        download_episodes: list[int] | None,
+        download_dir: Path,
+        torrent_content: bytes,
+    ) -> None:
+        """保持下载历史提交后的通知和后处理顺序。"""
+        self.post_message(
+            Message(
+                channel=channel,
+                source=source if channel else None,
+                mtype=MessageType.Download,
+                ctype=ContentType.DownloadAdded,
+                image=media.get_message_image(),
+                link=settings.MP_DOMAIN('/#/downloading'),
+                userid=userid,
+                username=username,
+            ),
+            meta=meta,
+            mediainfo=media,
+            torrentinfo=torrent,
+            download_episodes=download_episodes,
+            username=username,
+        )
+        self._submit_download_added_task(
+            context=context,
+            download_dir=download_dir,
+            torrent_content=torrent_content,
+        )
+
     @staticmethod
     def _validate_music_album_resource(
             context: Context,
@@ -1161,35 +1200,33 @@ class DownloadChain(ChainBase):
             # 文件保存路径
             _save_path = download_dir if _layout == "NoSubfolder" or not _folder_name else download_path
 
-            # 登记下载记录
-            downloadhis = DownloadHistoryOper()
             media_source, media_id = resolve_media_identity(media=_media)
-            downloadhis.add(
-                path=download_path.as_posix(),
-                type=_media.type.value,
-                title=_media.title,
-                year=_media.year,
-                media_source=media_source,
-                media_id=media_id,
-                music_type=getattr(_media, "music_type", None),
-                seasons=_meta.season,
-                episodes=download_episodes or _meta.episode,
-                image=_media.get_backdrop_image(),
-                poster=_media.get_poster_image(),
-                downloader=_downloader,
-                download_hash=_hash,
-                torrent_name=_torrent.title,
-                torrent_description=_torrent.description,
-                torrent_site=_torrent.site_name,
-                userid=userid,
-                username=username,
-                channel=channel.value if channel else None,
-                date=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
-                media_category=_media.category,
-                episode_group=_media.episode_group,
-                note=self._build_download_note(source, _media, _meta),
-                custom_words=custom_words
-            )
+            history_payload = {
+                "path": download_path.as_posix(),
+                "type": _media.type.value,
+                "title": _media.title,
+                "year": _media.year,
+                "media_source": media_source,
+                "media_id": media_id,
+                "music_type": getattr(_media, "music_type", None),
+                "seasons": _meta.season,
+                "episodes": download_episodes or _meta.episode,
+                "image": _media.get_backdrop_image(),
+                "poster": _media.get_poster_image(),
+                "downloader": _downloader,
+                "download_hash": _hash,
+                "torrent_name": _torrent.title,
+                "torrent_description": _torrent.description,
+                "torrent_site": _torrent.site_name,
+                "userid": userid,
+                "username": username,
+                "channel": channel.value if channel else None,
+                "date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
+                "media_category": _media.category,
+                "episode_group": _media.episode_group,
+                "note": self._build_download_note(source, _media, _meta),
+                "custom_words": custom_words,
+            }
 
             # 登记下载文件
             files_to_add = []
@@ -1213,42 +1250,51 @@ class DownloadChain(ChainBase):
                     "filepath": file,
                     "torrentname": _meta.org_string,
                 })
-            if files_to_add:
-                downloadhis.add_files(files_to_add)
-
-            # 下载成功发送消息
-            self.post_message(
-                Message(
-                    channel=channel,
-                    source=source if channel else None,
-                    mtype=MessageType.Download,
-                    ctype=ContentType.DownloadAdded,
-                    image=_media.get_message_image(),
-                    link=settings.MP_DOMAIN('/#/downloading'),
-                    userid=userid,
-                    username=username
-                ),
-                meta=_meta,
-                mediainfo=_media,
-                torrentinfo=_torrent,
-                download_episodes=download_episodes,
-                username=username,
-            )
-            # 下载成功后处理
-            self._submit_download_added_task(
-                context=context,
-                download_dir=download_dir,
-                torrent_content=torrent_content,
-            )
-            # 广播事件
-            self.eventmanager.send_event(EventType.DownloadAdded, {
+            event_payload = {
                 "hash": _hash,
                 "context": context,
                 "username": username,
                 "downloader": _downloader,
                 "episodes": episodes or _meta.episode_list,
-                "source": source
-            })
+                "source": source,
+            }
+
+            def after_commit() -> None:
+                """在历史与 intent 提交后保持原有通知和任务编排。"""
+                self._after_download_history_commit(
+                    context=context,
+                    media=_media,
+                    meta=_meta,
+                    torrent=_torrent,
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    download_episodes=download_episodes,
+                    download_dir=download_dir,
+                    torrent_content=torrent_content,
+                )
+
+            durable_event_writer = getattr(self, "durable_event_writer", None)
+            if durable_event_writer:
+                durable_event_writer.download_added(
+                    history_payload=history_payload,
+                    file_payloads=files_to_add,
+                    event_payload=event_payload,
+                    after_commit=after_commit,
+                    publish=lambda payload: self.eventmanager.send_event(
+                        EventType.DownloadAdded,
+                        payload,
+                    ),
+                )
+            else:
+                # 显式注入旧测试上下文时保持兼容；正式启动上下文总会提供 durable writer。
+                downloadhis = DownloadHistoryOper()
+                downloadhis.add(**history_payload)
+                if files_to_add:
+                    downloadhis.add_files(files_to_add)
+                after_commit()
+                self.eventmanager.send_event(EventType.DownloadAdded, event_payload)
         else:
             # 下载失败
             logger.error(f"{_media.title_year} 添加下载任务失败："

app/chain/transfer.py

@@ -87,6 +87,23 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo
         "TRANSFER_THREADS",
     }
 
+    @staticmethod
+    def _transfer_result_payload(
+        task: TransferTask,
+        transferinfo: TransferInfo,
+        history_id: int | None = None,
+    ) -> dict[str, Any]:
+        """构造保持插件旧对象字段不变的整理结果事件 payload。"""
+        return {
+            "fileitem": task.fileitem,
+            "meta": task.meta,
+            "mediainfo": task.mediainfo,
+            "transferinfo": transferinfo,
+            "downloader": task.downloader,
+            "download_hash": task.download_hash,
+            "transfer_history_id": history_id,
+        }
+
     def __init__(self):
         """初始化文件整理处理链。"""
         super().__init__()
@@ -229,33 +246,54 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo
                     fileid=task.fileitem.fileid if task.fileitem else None,
                 )
 
-                # 新增转移失败历史记录
-                history = add_transfer_fail(
-                    fileitem=task.fileitem,
-                    mode=transferinfo.transfer_type if transferinfo else "",
-                    downloader=task.downloader,
-                    download_hash=task.download_hash,
-                    meta=task.meta,
-                    mediainfo=task.mediainfo,
-                    transferinfo=transferinfo,
-                    transfer_history_oper=transferhis,
+                durable_transfer_failed = bool(
+                    getattr(self, "durable_event_writer", None)
+                    and self._is_media_file(task.fileitem)
                 )
+                if durable_transfer_failed:
+                    event_payload = self._transfer_result_payload(task, transferinfo)
+                    history = self.durable_event_writer.transfer_result(
+                        topic="transfer.failed",
+                        stage_history=lambda writer: add_transfer_fail(
+                            fileitem=task.fileitem,
+                            mode=transferinfo.transfer_type if transferinfo else "",
+                            downloader=task.downloader,
+                            download_hash=task.download_hash,
+                            meta=task.meta,
+                            mediainfo=task.mediainfo,
+                            transferinfo=transferinfo,
+                            transfer_history_oper=writer,
+                        ),
+                        event_payload=event_payload,
+                        publish=lambda payload: self.eventmanager.send_event(
+                            EventType.TransferFailed,
+                            payload,
+                        ),
+                    )
+                else:
+                    history = add_transfer_fail(
+                        fileitem=task.fileitem,
+                        mode=transferinfo.transfer_type if transferinfo else "",
+                        downloader=task.downloader,
+                        download_hash=task.download_hash,
+                        meta=task.meta,
+                        mediainfo=task.mediainfo,
+                        transferinfo=transferinfo,
+                        transfer_history_oper=transferhis,
+                    )
 
                 # 整理失败事件
                 if self._is_media_file(task.fileitem):
-                    # 主要媒体文件整理失败事件
-                    self.eventmanager.send_event(
-                        EventType.TransferFailed,
-                        {
-                            "fileitem": task.fileitem,
-                            "meta": task.meta,
-                            "mediainfo": task.mediainfo,
-                            "transferinfo": transferinfo,
-                            "downloader": task.downloader,
-                            "download_hash": task.download_hash,
-                            "transfer_history_id": history.id if history else None,
-                        },
-                    )
+                    if not durable_transfer_failed:
+                        # 显式旧测试上下文仍走原始发送；正式上下文由 outbox writer 发布。
+                        self.eventmanager.send_event(
+                            EventType.TransferFailed,
+                            self._transfer_result_payload(
+                                task,
+                                transferinfo,
+                                history.id if history else None,
+                            ),
+                        )
                 elif self._is_subtitle_file(task.fileitem):
                     # 字幕整理失败事件
                     self.eventmanager.send_event(
@@ -350,33 +388,54 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo
                 task.fileitem.storage if task.fileitem else None,
             )
 
-            # 新增task转移成功历史记录
-            history = add_transfer_success(
-                fileitem=task.fileitem,
-                mode=transferinfo.transfer_type if transferinfo else "",
-                downloader=task.downloader,
-                download_hash=task.download_hash,
-                meta=task.meta,
-                mediainfo=task.mediainfo,
-                transferinfo=transferinfo,
-                transfer_history_oper=transferhis,
+            durable_transfer_complete = bool(
+                getattr(self, "durable_event_writer", None)
+                and self._is_primary_media_file(task.fileitem, task.mediainfo)
             )
+            if durable_transfer_complete:
+                event_payload = self._transfer_result_payload(task, transferinfo)
+                history = self.durable_event_writer.transfer_result(
+                    topic="transfer.completed",
+                    stage_history=lambda writer: add_transfer_success(
+                        fileitem=task.fileitem,
+                        mode=transferinfo.transfer_type if transferinfo else "",
+                        downloader=task.downloader,
+                        download_hash=task.download_hash,
+                        meta=task.meta,
+                        mediainfo=task.mediainfo,
+                        transferinfo=transferinfo,
+                        transfer_history_oper=writer,
+                    ),
+                    event_payload=event_payload,
+                    publish=lambda payload: self.eventmanager.send_event(
+                        EventType.TransferComplete,
+                        payload,
+                    ),
+                )
+            else:
+                history = add_transfer_success(
+                    fileitem=task.fileitem,
+                    mode=transferinfo.transfer_type if transferinfo else "",
+                    downloader=task.downloader,
+                    download_hash=task.download_hash,
+                    meta=task.meta,
+                    mediainfo=task.mediainfo,
+                    transferinfo=transferinfo,
+                    transfer_history_oper=transferhis,
+                )
 
             # task整理完成事件
             if self._is_primary_media_file(task.fileitem, task.mediainfo):
-                # 主要媒体文件整理完成事件
-                self.eventmanager.send_event(
-                    EventType.TransferComplete,
-                    {
-                        "fileitem": task.fileitem,
-                        "meta": task.meta,
-                        "mediainfo": task.mediainfo,
-                        "transferinfo": transferinfo,
-                        "downloader": task.downloader,
-                        "download_hash": task.download_hash,
-                        "transfer_history_id": history.id if history else None,
-                    },
-                )
+                if not durable_transfer_complete:
+                    # 显式旧测试上下文仍走原始发送；正式上下文由 outbox writer 发布。
+                    self.eventmanager.send_event(
+                        EventType.TransferComplete,
+                        self._transfer_result_payload(
+                            task,
+                            transferinfo,
+                            history.id if history else None,
+                        ),
+                    )
             elif self._is_subtitle_file(task.fileitem):
                 # 字幕整理完成事件
                 self.eventmanager.send_event(

app/db/models/subscribe.py

@@ -1,12 +1,12 @@
 import time
 from typing import Any, Optional
 
-from sqlalchemy import Integer, String, Float, JSON, Index, delete, or_, select
+from sqlalchemy import Integer, String, Float, JSON, Index, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import get_id_column, Base
-from app.db.decorators import db_query, db_update, async_db_query, async_db_update
+from app.db.decorators import async_db_query, db_query
 from app.db.models._constraints import media_identity_constraint
 from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource
 
@@ -363,36 +363,6 @@ class Subscribe(Base):
         result = await db.execute(query)
         return result.scalars().first()
 
-    @db_update
-    def delete_by_media_identity(
-            self, db: Session, media_source: MediaSource, media_id: str,
-            season: Optional[int] = None,
-    ) -> bool:
-        """按规范媒体身份删除订阅。"""
-        model = type(self)
-        statement = delete(model).where(
-            model.media_source == media_source,
-            model.media_id == str(media_id),
-        )
-        if season is not None:
-            statement = statement.where(model.season == season)
-        db.execute(statement, execution_options={"synchronize_session": False})
-        return True
-
-    @async_db_update
-    async def async_delete_by_media_identity(
-            self, db: AsyncSession, media_source: MediaSource, media_id: str,
-            season: Optional[int] = None,
-    ) -> bool:
-        """异步按规范媒体身份删除订阅。"""
-        rows = await self.async_list_by_media_identity(
-            db, media_source=media_source, media_id=media_id
-        )
-        for row in rows:
-            if season is None or row.season == season:
-                await row.async_delete(db, row.id)
-        return True
-
     @classmethod
     @db_query
     def list_by_username(cls, db: Session, username: str, state: Optional[str] = None, mtype: Optional[str] = None):

app/db/oper/downloadhistory.py

@@ -1,6 +1,7 @@
 from typing import Dict, List, Optional, cast
 
 from sqlalchemy import delete as sqlalchemy_delete, update as sqlalchemy_update
+from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
 from app.db.models.downloadhistory import DownloadHistory, DownloadFiles
@@ -60,6 +61,15 @@ class DownloadHistoryOper(DbOper):
         """
         DownloadHistory(**kwargs).create(self._db)
 
+    def stage_add(self, payload: dict) -> DownloadHistory:
+        """在调用方同步 Session 中暂存下载历史并返回已分配 ID 的记录。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("下载历史事务写入需要调用方提供同步 Session")
+        history = DownloadHistory(**payload)
+        self._db.add(history)
+        self._db.flush()
+        return history
+
     def add_files(self, file_items: List[dict]):
         """
         新增下载历史文件
@@ -68,6 +78,13 @@ class DownloadHistoryOper(DbOper):
             downloadfile = DownloadFiles(**file_item)
             downloadfile.create(self._db)
 
+    def stage_add_files(self, file_items: List[dict]) -> None:
+        """在调用方事务内批量暂存下载文件，不逐条提交。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("下载文件事务写入需要调用方提供同步 Session")
+        self._db.add_all(DownloadFiles(**item) for item in file_items)
+        self._db.flush()
+
     def truncate_files(self):
         """
         清空下载历史文件记录

app/db/oper/transferhistory.py

@@ -2,6 +2,7 @@ import time
 from typing import Any, List, Optional
 
 from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
 from app.db.models.transferhistory import TransferHistory
@@ -272,6 +273,24 @@ class TransferHistoryOper(DbOper):
             kwargs["src_storage"],
         )
 
+    def stage_replace_by_src(self, **kwargs) -> TransferHistory:
+        """在调用方事务内按源路径替换整理历史并返回已分配 ID 的新记录。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("整理历史事务写入需要调用方提供同步 Session")
+        kwargs["src_storage"] = kwargs.get("src_storage") or "local"
+        kwargs["date"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+        self._db.execute(
+            sqlalchemy_delete(TransferHistory).where(
+                TransferHistory.src == kwargs.get("src"),
+                TransferHistory.src_storage == kwargs["src_storage"],
+            )
+        )
+        self._db.flush()
+        history = TransferHistory(**kwargs)
+        self._db.add(history)
+        self._db.flush()
+        return history
+
     def update_download_hash(self, historyid, download_hash):
         """
         补充转移记录download_hash

app/runtime/event/contracts.py

@@ -57,6 +57,9 @@ _PAYLOAD_MODELS: dict[EventType | ChainEventType, type[BaseModel]] = {
     EventType.SubscribeAdded: event_schemas.SubscribeAddedEventData,
     EventType.SubscribeDeleted: event_schemas.SubscribeDeletedEventData,
     EventType.SubscribeModified: event_schemas.SubscribeModifiedEventData,
+    EventType.DownloadAdded: event_schemas.DownloadAddedEventData,
+    EventType.TransferComplete: event_schemas.TransferResultEventData,
+    EventType.TransferFailed: event_schemas.TransferResultEventData,
     ChainEventType.PluginDataReset: event_schemas.PluginDataResetEventData,
     ChainEventType.AuthVerification: event_schemas.AuthCredentials,
     ChainEventType.AuthIntercept: event_schemas.AuthInterceptCredentials,

app/schemas/event.py

@@ -7,6 +7,7 @@ from app.schemas.common import JsonData
 from app.schemas.types import MediaType, NotificationChannel
 from app.schemas.file import FileItem
 from app.schemas.media import OptionalMediaIdentityMixin, RequiredMediaIdentityMixin
+from app.schemas.transfer import TransferInfo
 from app.schemas.types import MediaSource
 
 
@@ -730,6 +731,31 @@ class SubscribeDeletedEventData(BaseEventData):
     idempotency_key: Optional[str] = Field(default=None, description="宿主生成的幂等键")
 
 
+class DownloadAddedEventData(BaseEventData):
+    """DownloadAdded 广播事件的插件兼容 payload。"""
+
+    hash: str = Field(description="下载任务 hash")
+    context: Any = Field(description="下载上下文对象")
+    username: Optional[str] = Field(default=None, description="发起下载的用户")
+    downloader: Optional[str] = Field(default=None, description="下载器名称")
+    episodes: List[int] = Field(default_factory=list, description="下载剧集列表")
+    source: Optional[str] = Field(default=None, description="下载来源")
+    idempotency_key: Optional[str] = Field(default=None, description="宿主生成的幂等键")
+
+
+class TransferResultEventData(BaseEventData):
+    """TransferComplete/Failed 共用的插件兼容 payload。"""
+
+    fileitem: Optional[FileItem] = Field(default=None, description="源文件项")
+    meta: Any = Field(default=None, description="文件名解析对象")
+    mediainfo: Any = Field(default=None, description="媒体信息对象")
+    transferinfo: Optional[TransferInfo] = Field(default=None, description="整理结果")
+    downloader: Optional[str] = Field(default=None, description="下载器名称")
+    download_hash: Optional[str] = Field(default=None, description="下载任务 hash")
+    transfer_history_id: Optional[int] = Field(default=None, description="整理历史 ID")
+    idempotency_key: Optional[str] = Field(default=None, description="宿主生成的幂等键")
+
+
 class SubscribeCompletionCheckEventData(ChainEventData):
     """
     SubscribeCompletionCheck 事件的数据模型

app/schemas/exports.py

@@ -85,6 +85,7 @@ SCHEMA_EXPORTS = {
     'DiscoverSourceEventData': ('app.schemas.event', 'DiscoverSourceEventData'),
     'Discriminator': ('app.schemas.context', 'Discriminator'),
     'DownloadAddedData': ('app.schemas.download', 'DownloadAddedData'),
+    'DownloadAddedEventData': ('app.schemas.event', 'DownloadAddedEventData'),
     'DownloadDirectory': ('app.schemas.download', 'DownloadDirectory'),
     'DownloadHistory': ('app.schemas.history', 'DownloadHistory'),
     'DownloadTask': ('app.schemas.workflow', 'DownloadTask'),
@@ -369,6 +370,7 @@ SCHEMA_EXPORTS = {
     'TransferOverwriteCheckEventData': ('app.schemas.event', 'TransferOverwriteCheckEventData'),
     'TransferRenameBuildEventData': ('app.schemas.event', 'TransferRenameBuildEventData'),
     'TransferRenameEventData': ('app.schemas.event', 'TransferRenameEventData'),
+    'TransferResultEventData': ('app.schemas.event', 'TransferResultEventData'),
     'TransferTorrent': ('app.schemas.transfer', 'TransferTorrent'),
     'TypeAdapter': ('app.schemas.mcp', 'TypeAdapter'),
     'TypeAlias': ('app.schemas.mcp', 'TypeAlias'),
@@ -441,6 +443,7 @@ SCHEMA_CONFLICTS = {
     'Subscribe': ['app.schemas.subscribe', 'app.schemas.workflow'],
     'SubtitleInfo': ['app.schemas.context', 'app.schemas.search'],
     'TorrentInfo': ['app.schemas.context', 'app.schemas.search', 'app.schemas.system'],
+    'TransferInfo': ['app.schemas.event', 'app.schemas.transfer'],
     'Union': ['app.schemas.agent', 'app.schemas.common', 'app.schemas.context', 'app.schemas.mediaserver', 'app.schemas.message', 'app.schemas.plugin', 'app.schemas.search', 'app.schemas.site', 'app.schemas.transfer', 'app.schemas.mcp'],
     'UserPermissions': ['app.schemas.token', 'app.schemas.user'],
     'dataclass': ['app.schemas.notification', 'app.schemas.system'],

app/startup/chain_events.py

@@ -0,0 +1,152 @@
+"""Chain durable 事件写入端口的 SQLAlchemy 启动适配器。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from app.application.chain.durable_events import (
+    ChainDurableEventWriter,
+    TransferHistoryRef,
+    download_added_event_key,
+    snapshot_download_added,
+    snapshot_transfer_result,
+    transfer_result_event_key,
+)
+from app.application.history import TransferHistoryRecord, TransferHistoryWriter
+from app.application.outbox import DurableEventCommand, OutboxIntent
+from app.db.oper.downloadhistory import DownloadHistoryOper
+from app.db.oper.transferhistory import TransferHistoryOper
+from app.db.uow import SqlAlchemyUnitOfWork
+from app.startup.outbox import SqlAlchemyOutboxRepository
+
+
+class _StagingTransferHistoryWriter:
+    """让既有历史字段映射复用无提交的 replace 适配器。"""
+
+    def __init__(self, repository: TransferHistoryOper) -> None:
+        """保存绑定调用方 Session 的整理历史仓储。"""
+        self._repository = repository
+
+    def get_by_src(
+        self,
+        src: str,
+        storage: str | None = None,
+    ) -> TransferHistoryRecord | None:
+        """转发按源路径读取。"""
+        return self._repository.get_by_src(src, storage)
+
+    def get_success_by_src(
+        self,
+        src: str,
+        storage: str | None = None,
+    ) -> TransferHistoryRecord | None:
+        """转发按源路径读取成功记录。"""
+        return self._repository.get_success_by_src(src, storage)
+
+    def add_force(self, **payload: Any) -> TransferHistoryRecord:
+        """保持应用层旧端口名，但只暂存替换而不自行提交。"""
+        return self._repository.stage_replace_by_src(**payload)
+
+
+class TransactionalChainDurableEventWriter(ChainDurableEventWriter):
+    """为每次 Chain 结果事件创建独占同步 Session 和 UoW。"""
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """注入惰性同步 Session 工厂。"""
+        self._session_factory = session_factory
+
+    def download_added(
+        self,
+        *,
+        history_payload: dict[str, Any],
+        file_payloads: list[dict[str, Any]],
+        event_payload: dict[str, Any],
+        after_commit: Callable[[], None],
+        publish: Callable[[dict[str, Any]], None],
+    ) -> None:
+        """原子写下载历史、文件清单和 DownloadAdded intent。"""
+        session = self._session_factory()
+        try:
+            repository = DownloadHistoryOper(session)
+            outbox = SqlAlchemyOutboxRepository(session)
+            command = DurableEventCommand(
+                unit_of_work=SqlAlchemyUnitOfWork(session),
+                outbox=outbox,
+            )
+            event_key = download_added_event_key(event_payload)
+            event_payload["idempotency_key"] = event_key
+
+            def stage_business() -> None:
+                """在同一事务暂存下载历史和可选文件清单。"""
+                repository.stage_add(history_payload)
+                if file_payloads:
+                    repository.stage_add_files(file_payloads)
+
+            command.execute(
+                intent=OutboxIntent(
+                    event_key=event_key,
+                    topic="download.added",
+                    payload=snapshot_download_added(event_payload),
+                ),
+                stage_business=stage_business,
+                after_commit=after_commit,
+                publish=lambda: publish(event_payload),
+            )
+        finally:
+            session.close()
+
+    def transfer_result(
+        self,
+        *,
+        topic: str,
+        stage_history: Callable[[TransferHistoryWriter], TransferHistoryRecord | None],
+        event_payload: dict[str, Any],
+        publish: Callable[[dict[str, Any]], None],
+    ) -> TransferHistoryRecord | None:
+        """原子写整理历史与结果 intent，并返回脱离 Session 的最小投影。"""
+        session = self._session_factory()
+        try:
+            staging = _StagingTransferHistoryWriter(TransferHistoryOper(session))
+            command = DurableEventCommand(
+                unit_of_work=SqlAlchemyUnitOfWork(session),
+                outbox=SqlAlchemyOutboxRepository(session),
+            )
+
+            def stage_business() -> TransferHistoryRef | None:
+                """复用历史字段映射，并在 flush 后冻结安全投影。"""
+                history = stage_history(staging)
+                if history is None:
+                    return None
+                return TransferHistoryRef(
+                    id=history.id,
+                    status=bool(history.status),
+                    src=history.src,
+                    src_storage=history.src_storage,
+                    src_fileitem=history.src_fileitem,
+                )
+
+            def build_intent(
+                history: TransferHistoryRef | None,
+            ) -> OutboxIntent:
+                """历史 ID 确定后构造事件键与可恢复快照。"""
+                if history is None:
+                    raise RuntimeError("整理历史暂存失败，无法登记 durable 结果事件")
+                event_key = transfer_result_event_key(topic, history.id)
+                event_payload["transfer_history_id"] = history.id
+                event_payload["idempotency_key"] = event_key
+                return OutboxIntent(
+                    event_key=event_key,
+                    topic=topic,
+                    payload=snapshot_transfer_result(event_payload),
+                )
+
+            return command.execute(
+                intent=build_intent,
+                stage_business=stage_business,
+                publish=lambda: publish(event_payload),
+            )
+        finally:
+            session.close()

app/startup/modules_initializer.py

@@ -107,6 +107,7 @@ from app.startup.subscription import (
     TransactionalSubscribeWriter,
     configure_transactional_subscription_scopes,
 )
+from app.startup.chain_events import TransactionalChainDurableEventWriter
 from app.startup.context import AgentChatRuntime, HostRuntime, SubscriptionRuntime
 from app.adapters.web.security.access import set_superuser_token_payload_provider
 from app.application.security.auth import build_superuser_token_payload
@@ -115,6 +116,10 @@ from app.application.chain.context import (
     ChainRuntimeContext,
     configure_chain_runtime_context_provider,
 )
+from app.application.chain.durable_events import (
+    restore_download_added,
+    restore_transfer_result,
+)
 from app.application.chain.data import configure_chain_data_ports, get_chain_data_ports
 from app.runtime.extensions.service_config import (
     ServiceConfigHelper,
@@ -147,6 +152,7 @@ def _build_chain_runtime_context() -> ChainRuntimeContext:
         ),
         module_dispatcher_factory=ModuleInvocationDispatcher,
         data_ports=get_chain_data_ports(),
+        durable_event_writer=TransactionalChainDurableEventWriter(SessionFactory),
     )
 
 
@@ -217,6 +223,18 @@ def _build_outbox_dispatcher() -> OutboxDispatcher:
                 EventType.SubscribeDeleted,
                 message.payload,
             ),
+            "download.added": lambda message: EventManager().send_event(
+                EventType.DownloadAdded,
+                restore_download_added(message.payload),
+            ),
+            "transfer.completed": lambda message: EventManager().send_event(
+                EventType.TransferComplete,
+                restore_transfer_result(message.payload),
+            ),
+            "transfer.failed": lambda message: EventManager().send_event(
+                EventType.TransferFailed,
+                restore_transfer_result(message.payload),
+            ),
         },
         close=session.close,
     )

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` 广播已由
-业务事务内的 outbox intent 提供 at-least-once 恢复；payload 保持插件 dict ABI，并增加可选幂等键。
-这不覆盖第三方插件自行发送的裸事件，也不代表订阅通知和外部统计上报已经全部 durable。
+截至 2026-08-22，宿主正式装配的 `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`、
+`DownloadAdded`、`TransferComplete`、`TransferFailed` 广播已由业务事务内的 outbox intent 提供
+at-least-once 恢复；payload 保持插件 dict/对象 ABI，并增加可选幂等键。下载和整理的 outbox 只保存
+可 JSON 序列化的快照，重放时恢复旧对象字段。这不覆盖第三方插件自行发送的裸事件，也不代表订阅通知
+和外部统计上报已经全部 durable。
 
 ## Event 映射
 
@@ -97,6 +99,6 @@ pending 状态交由下次启动，不以取消异常写成成功。
 ## 验证与演进
 
 - Event Registry 的 `delivery` 字段与本 ADR 同步进入 runtime baseline。
-- ARCH-251 从一个 E2 pilot 扩展到三种订阅生命周期事件，并通过 commit 后崩溃、重复 claim、并发 claim
-  和 dead-letter 测试；下载与整理结果事件仍需逐条迁移。
+- ARCH-251 已覆盖 Registry 中六种 `durable_required` 事件，并通过 commit 后崩溃、重复 claim、并发
+  claim、JSON 快照恢复和 dead-letter 测试；后续新增 E2 事件必须同时提供业务事务边界和恢复测试。
 - ARCH-252 将 Scheduler 的定义、触发和执行状态拆分，但不提升不需要 durable 的 E0 信号。

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

@@ -743,6 +743,16 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas
   插件若自行直接发送同名事件，该发送仍由插件负责，无法与插件自己的数据库写入自动组成原子事务。
 - 订阅外部统计上报仍是 post-commit 副作用，不在事件 intent 的重放 handler 中；因此当前可以宣称三种
   订阅事件具备宿主级 at-least-once 恢复，但不能宣称订阅通知和所有外部上报均已 durable。
+- `DownloadAdded`、`TransferComplete`、`TransferFailed` 也已逐项接入，而不是复用一个不分业务语义的
+  “万能消息总线”。下载历史、下载文件清单或整理历史与各自 intent 在独占同步 Session/UoW 中原子提交；
+  即时广播失败时 intent 保持 pending，三种恢复 handler 均继续使用有限重试与 dead-letter 策略。
+- 下载和整理事件保留插件原有运行时对象 ABI：即时发送仍含 `Context`、`FileItem`、`MetaInfo`、
+  `MediaInfo`、`TransferInfo`；outbox 单独存 JSON 快照，恢复 handler 无远端调用地重建这些对象。
+  `idempotency_key` 仍是唯一新增的可选公开字段，提醒插件按 at-least-once 语义自行去重。
+- 本切片同时把 `DownloadChain.download_single` 的提交后通知/任务编排抽成独立方法，并删除已经被
+  Application 删除命令替代的两个 `Subscribe` Model 级删除事务装饰器；Model decorator 基线从
+  178 降到 176，Oper 内显式 commit/rollback 仍为 0。strict mypy 门禁新增 Chain durable context、
+  payload 转换和启动适配器。
 
 **禁止**：本阶段不引入 Celery、Kafka、RabbitMQ 等新基础设施。
 

mypy.ini

@@ -16,9 +16,12 @@ files =
     app/runtime/event/contracts.py,
     app/runtime/extensions/module/contracts.py,
     app/application/outbox.py,
+    app/application/chain/context.py,
+    app/application/chain/durable_events.py,
     app/application/subscription/delete.py,
     app/application/subscription/identity.py,
     app/application/subscription/mutation.py,
     app/startup/context.py,
+    app/startup/chain_events.py,
     app/api/context.py,
     app/api/dependencies/subscription.py

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6271,
-  "edge_sha256": "0119519add8e4499684044fc69f3003c54cfca20bf04f4614a419374692d5a94",
+  "edge_count": 6299,
+  "edge_sha256": "f98ab7d05c884a07a881eff1ab34b1ba5023751f1e95b956debcbad54c385810",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -2438,6 +2438,19 @@
     "app.application.chain.context -> app.application",
     "app.application.chain.context -> app.application.chain",
     "app.application.chain.context -> app.application.chain.data",
+    "app.application.chain.context -> app.application.chain.durable_events",
+    "app.application.chain.durable_events -> app.application",
+    "app.application.chain.durable_events -> app.application.history",
+    "app.application.chain.durable_events -> app.domain",
+    "app.application.chain.durable_events -> app.domain.context",
+    "app.application.chain.durable_events -> app.domain.meta",
+    "app.application.chain.durable_events -> app.domain.meta.metabase",
+    "app.application.chain.durable_events -> app.domain.meta.metamusic",
+    "app.application.chain.durable_events -> app.domain.metainfo",
+    "app.application.chain.durable_events -> app.schemas",
+    "app.application.chain.durable_events -> app.schemas.file",
+    "app.application.chain.durable_events -> app.schemas.transfer",
+    "app.application.chain.durable_events -> app.schemas.types",
     "app.application.dashboard -> app.schemas",
     "app.application.dashboard -> app.schemas.dashboard",
     "app.application.database -> app.application",
@@ -5623,6 +5636,7 @@
     "app.schemas.event -> app.schemas.common",
     "app.schemas.event -> app.schemas.file",
     "app.schemas.event -> app.schemas.media",
+    "app.schemas.event -> app.schemas.transfer",
     "app.schemas.event -> app.schemas.types",
     "app.schemas.file -> app.schemas",
     "app.schemas.file -> app.schemas.types",
@@ -5841,6 +5855,18 @@
     "app.startup.cache_initializer -> app.adapters",
     "app.startup.cache_initializer -> app.adapters.cache",
     "app.startup.cache_initializer -> app.adapters.cache.backends",
+    "app.startup.chain_events -> app.application",
+    "app.startup.chain_events -> app.application.chain",
+    "app.startup.chain_events -> app.application.chain.durable_events",
+    "app.startup.chain_events -> app.application.history",
+    "app.startup.chain_events -> app.application.outbox",
+    "app.startup.chain_events -> app.db",
+    "app.startup.chain_events -> app.db.oper",
+    "app.startup.chain_events -> app.db.oper.downloadhistory",
+    "app.startup.chain_events -> app.db.oper.transferhistory",
+    "app.startup.chain_events -> app.db.uow",
+    "app.startup.chain_events -> app.startup",
+    "app.startup.chain_events -> app.startup.outbox",
     "app.startup.command_initializer -> app.application",
     "app.startup.command_initializer -> app.application.commands",
     "app.startup.command_initializer -> app.command",
@@ -5950,6 +5976,7 @@
     "app.startup.modules_initializer -> app.application.chain",
     "app.startup.modules_initializer -> app.application.chain.context",
     "app.startup.modules_initializer -> app.application.chain.data",
+    "app.startup.modules_initializer -> app.application.chain.durable_events",
     "app.startup.modules_initializer -> app.application.configuration",
     "app.startup.modules_initializer -> app.application.database",
     "app.startup.modules_initializer -> app.application.history",
@@ -6026,6 +6053,7 @@
     "app.startup.modules_initializer -> app.schemas.types",
     "app.startup.modules_initializer -> app.startup",
     "app.startup.modules_initializer -> app.startup.agent_initializer",
+    "app.startup.modules_initializer -> app.startup.chain_events",
     "app.startup.modules_initializer -> app.startup.context",
     "app.startup.modules_initializer -> app.startup.database",
     "app.startup.modules_initializer -> app.startup.managed_resources_initializer",
@@ -6288,7 +6316,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 782,
+  "module_count": 784,
   "modules": [
     "app",
     "app.adapters",
@@ -6542,6 +6570,7 @@
     "app.application.chain",
     "app.application.chain.context",
     "app.application.chain.data",
+    "app.application.chain.durable_events",
     "app.application.commands",
     "app.application.configuration",
     "app.application.dashboard",
@@ -7034,6 +7063,7 @@
     "app.startup",
     "app.startup.agent_initializer",
     "app.startup.cache_initializer",
+    "app.startup.chain_events",
     "app.startup.command_initializer",
     "app.startup.context",
     "app.startup.database",

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

@@ -1441,10 +1441,10 @@
     "EventType.DownloadAdded": {
       "delivery": "durable_required",
       "error_behavior": "notify",
-      "legacy_reason": "现有插件 payload 尚未收敛为稳定 model，保留原始 dict ABI",
+      "legacy_reason": null,
       "mode": "broadcast",
       "ordering": "priority_queue",
-      "payload_contract": "legacy_dict",
+      "payload_contract": "DownloadAddedEventData",
       "sensitive_fields": [],
       "visibility": "plugin_public"
     },
@@ -1651,20 +1651,20 @@
     "EventType.TransferComplete": {
       "delivery": "durable_required",
       "error_behavior": "notify",
-      "legacy_reason": "现有插件 payload 尚未收敛为稳定 model，保留原始 dict ABI",
+      "legacy_reason": null,
       "mode": "broadcast",
       "ordering": "priority_queue",
-      "payload_contract": "legacy_dict",
+      "payload_contract": "TransferResultEventData",
       "sensitive_fields": [],
       "visibility": "plugin_public"
     },
     "EventType.TransferFailed": {
       "delivery": "durable_required",
       "error_behavior": "notify",
-      "legacy_reason": "现有插件 payload 尚未收敛为稳定 model，保留原始 dict ABI",
+      "legacy_reason": null,
       "mode": "broadcast",
       "ordering": "priority_queue",
-      "payload_contract": "legacy_dict",
+      "payload_contract": "TransferResultEventData",
       "sensitive_fields": [],
       "visibility": "plugin_public"
     },
@@ -2032,6 +2032,10 @@
         "producers": [
           {
             "caller": "app.chain.download",
+            "count": 2
+          },
+          {
+            "caller": "app.startup.modules_initializer",
             "count": 1
           }
         ]
@@ -2274,6 +2278,10 @@
         "producers": [
           {
             "caller": "app.chain.transfer",
+            "count": 2
+          },
+          {
+            "caller": "app.startup.modules_initializer",
             "count": 1
           }
         ]
@@ -2283,6 +2291,10 @@
         "producers": [
           {
             "caller": "app.chain.transfer",
+            "count": 2
+          },
+          {
+            "caller": "app.startup.modules_initializer",
             "count": 1
           }
         ]
@@ -2320,7 +2332,7 @@
         ]
       }
     },
-    "producer_count": 70
+    "producer_count": 76
   },
   "module_method_specs": {
     "download_file": {

tests/fixtures/architecture/transaction-debt-baseline.json

@@ -2,11 +2,11 @@
   "model_decorators": {
     "by_kind": {
       "async_db_query": 49,
-      "async_db_update": 13,
+      "async_db_update": 12,
       "db_query": 75,
-      "db_update": 41
+      "db_update": 40
     },
-    "count": 178,
+    "count": 176,
     "methods": [
       {
         "decorator": "async_db_query",
@@ -463,11 +463,6 @@
         "file": "app/db/models/siteuserdata.py",
         "method": "SiteUserData.get_latest"
       },
-      {
-        "decorator": "async_db_update",
-        "file": "app/db/models/subscribe.py",
-        "method": "Subscribe.async_delete_by_media_identity"
-      },
       {
         "decorator": "async_db_query",
         "file": "app/db/models/subscribe.py",
@@ -513,11 +508,6 @@
         "file": "app/db/models/subscribe.py",
         "method": "Subscribe.async_list_by_username"
       },
-      {
-        "decorator": "db_update",
-        "file": "app/db/models/subscribe.py",
-        "method": "Subscribe.delete_by_media_identity"
-      },
       {
         "decorator": "db_query",
         "file": "app/db/models/subscribe.py",

tests/test_architecture_contract_baseline.py

@@ -125,8 +125,8 @@ def test_transaction_debt_baseline_is_a_model_and_oper_ratchet() -> None:
     baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
 
     assert baseline["schema_version"] == 1
-    assert baseline["model_decorators"]["count"] == 178
-    assert sum(baseline["model_decorators"]["by_kind"].values()) == 178
+    assert baseline["model_decorators"]["count"] == 176
+    assert sum(baseline["model_decorators"]["by_kind"].values()) == 176
     assert baseline["model_transaction_calls"] == {"count": 0, "calls": []}
     assert baseline["model_session_factories"] == {"count": 0, "calls": []}
     assert baseline["oper_transaction_calls"] == {"count": 0, "calls": []}

tests/test_chain_durable_events.py

@@ -0,0 +1,219 @@
+"""下载与整理 durable 事件的原子写入和对象恢复测试。"""
+
+import json
+
+import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import sessionmaker
+
+from app.application.chain.durable_events import (
+    restore_download_added,
+    restore_transfer_result,
+    snapshot_download_added,
+    snapshot_transfer_result,
+)
+from app.db.base import Base
+from app.db.models.downloadhistory import DownloadFiles, DownloadHistory
+from app.db.models.outbox import OutboxMessage
+from app.db.models.transferhistory import TransferHistory
+from app.domain.context import Context, MediaInfo, TorrentInfo
+from app.domain.metainfo import MetaInfo
+from app.schemas.file import FileItem
+from app.schemas.transfer import TransferInfo
+from app.schemas.types import MediaSource, MediaType
+from app.startup.chain_events import TransactionalChainDurableEventWriter
+
+
+def _session_factory():
+    """创建只服务当前测试的内存数据库和同步 Session 工厂。"""
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    return sessionmaker(bind=engine)
+
+
+def _objects():
+    """构造下载和整理事件共用的最小真实领域对象。"""
+    meta = MetaInfo("Demo.2026.1080p.mkv")
+    media = MediaInfo(
+        type=MediaType.MOVIE,
+        title="Demo",
+        year="2026",
+        media_source=MediaSource.TMDB,
+        media_id="42",
+    )
+    torrent = TorrentInfo(title="Demo torrent", site_name="test")
+    context = Context(meta_info=meta, media_info=media, torrent_info=torrent)
+    fileitem = FileItem(
+        storage="local",
+        path="/downloads/Demo.mkv",
+        name="Demo.mkv",
+        type="file",
+    )
+    transferinfo = TransferInfo(
+        success=True,
+        fileitem=fileitem,
+        target_item=FileItem(
+            storage="local",
+            path="/library/Demo (2026)/Demo.mkv",
+            name="Demo.mkv",
+            type="file",
+        ),
+        transfer_type="copy",
+    )
+    return meta, media, context, fileitem, transferinfo
+
+
+def test_durable_snapshots_are_json_and_restore_plugin_runtime_objects():
+    """outbox 只存 JSON 快照，重放时恢复插件一直收到的对象类型。"""
+    meta, media, context, fileitem, transferinfo = _objects()
+    download_payload = {
+        "hash": "hash-1",
+        "context": context,
+        "username": "alice",
+        "downloader": "qb",
+        "episodes": [1, 2],
+        "source": "manual",
+        "idempotency_key": "download.added:qb:hash-1:v1",
+    }
+    transfer_payload = {
+        "fileitem": fileitem,
+        "meta": meta,
+        "mediainfo": media,
+        "transferinfo": transferinfo,
+        "downloader": "qb",
+        "download_hash": "hash-1",
+        "transfer_history_id": 9,
+        "idempotency_key": "transfer.completed:9:v1",
+    }
+
+    download_snapshot = snapshot_download_added(download_payload)
+    transfer_snapshot = snapshot_transfer_result(transfer_payload)
+    json.dumps(download_snapshot)
+    json.dumps(transfer_snapshot)
+
+    restored_download = restore_download_added(download_snapshot)
+    restored_transfer = restore_transfer_result(transfer_snapshot)
+    assert isinstance(restored_download["context"], Context)
+    assert isinstance(restored_download["context"].media_info, MediaInfo)
+    assert isinstance(restored_transfer["fileitem"], FileItem)
+    assert type(restored_transfer["meta"]) is type(meta)
+    assert isinstance(restored_transfer["mediainfo"], MediaInfo)
+    assert isinstance(restored_transfer["transferinfo"], TransferInfo)
+
+
+def test_download_history_and_event_intent_share_one_transaction():
+    """下载历史、文件清单和 intent 提交后才执行通知与即时事件。"""
+    factory = _session_factory()
+    writer = TransactionalChainDurableEventWriter(factory)
+    _, _, context, _, _ = _objects()
+    calls = []
+
+    writer.download_added(
+        history_payload={
+            "path": "/downloads/Demo.mkv",
+            "type": MediaType.MOVIE.value,
+            "title": "Demo",
+            "download_hash": "hash-2",
+        },
+        file_payloads=[
+            {
+                "download_hash": "hash-2",
+                "downloader": "qb",
+                "fullpath": "/downloads/Demo.mkv",
+                "savepath": "/downloads",
+                "filepath": "Demo.mkv",
+                "torrentname": "Demo torrent",
+            }
+        ],
+        event_payload={
+            "hash": "hash-2",
+            "context": context,
+            "username": "alice",
+            "downloader": "qb",
+            "episodes": [],
+            "source": "manual",
+        },
+        after_commit=lambda: calls.append("after_commit"),
+        publish=lambda payload: calls.append(("event", payload)),
+    )
+
+    with factory() as session:
+        history = session.execute(select(DownloadHistory)).scalar_one()
+        download_file = session.execute(select(DownloadFiles)).scalar_one()
+        outbox = session.execute(select(OutboxMessage)).scalar_one()
+    assert history.download_hash == "hash-2"
+    assert download_file.fullpath == "/downloads/Demo.mkv"
+    assert outbox.status == "completed"
+    assert outbox.event_key == "download.added:qb:hash-2:v1"
+    assert [call if isinstance(call, str) else call[0] for call in calls] == [
+        "after_commit",
+        "event",
+    ]
+
+    with pytest.raises(IntegrityError):
+        writer.download_added(
+            history_payload={
+                "path": "/downloads/duplicate.mkv",
+                "type": MediaType.MOVIE.value,
+                "title": "Duplicate",
+                "download_hash": "hash-2",
+            },
+            file_payloads=[],
+            event_payload={
+                "hash": "hash-2",
+                "context": context,
+                "downloader": "qb",
+                "episodes": [],
+            },
+            after_commit=lambda: None,
+            publish=lambda _payload: None,
+        )
+    with factory() as session:
+        assert len(session.execute(select(DownloadHistory)).scalars().all()) == 1
+
+
+def test_transfer_event_failure_leaves_committed_intent_pending():
+    """整理历史提交后即时广播失败不得误删可供恢复的 pending intent。"""
+    factory = _session_factory()
+    writer = TransactionalChainDurableEventWriter(factory)
+    meta, media, _, fileitem, transferinfo = _objects()
+
+    def stage_history(repository):
+        """通过应用历史端口名暂存一条成功整理记录。"""
+        return repository.add_force(
+            src=fileitem.path,
+            src_storage=fileitem.storage,
+            src_fileitem=fileitem.model_dump(mode="json"),
+            dest=transferinfo.target_item.path,
+            dest_storage=transferinfo.target_item.storage,
+            dest_fileitem=transferinfo.target_item.model_dump(mode="json"),
+            status=1,
+        )
+
+    def fail_publish(_payload):
+        """模拟插件事件总线在业务提交后失败。"""
+        raise RuntimeError("event failed")
+
+    with pytest.raises(RuntimeError, match="event failed"):
+        writer.transfer_result(
+            topic="transfer.completed",
+            stage_history=stage_history,
+            event_payload={
+                "fileitem": fileitem,
+                "meta": meta,
+                "mediainfo": media,
+                "transferinfo": transferinfo,
+                "downloader": "qb",
+                "download_hash": "hash-3",
+                "transfer_history_id": None,
+            },
+            publish=fail_publish,
+        )
+
+    with factory() as session:
+        history = session.execute(select(TransferHistory)).scalar_one()
+        outbox = session.execute(select(OutboxMessage)).scalar_one()
+    assert history.status is True
+    assert outbox.status == "pending"
+    assert outbox.event_key == f"transfer.completed:{history.id}:v1"

tests/test_chain_runtime_context.py

@@ -20,6 +20,7 @@ def _context() -> ChainRuntimeContext:
         async_file_cache=Mock(),
         message_queue_factory=Mock(return_value=Mock()),
         module_dispatcher_factory=ModuleInvocationDispatcher,
+        durable_event_writer=Mock(),
     )
 
 
@@ -33,6 +34,7 @@ def test_chain_accepts_explicit_runtime_context() -> None:
     assert chain.pluginmanager is context.plugin_manager
     assert chain.eventmanager is context.event_manager
     assert chain.messagehelper is context.message_helper
+    assert chain.durable_event_writer is context.durable_event_writer
     context.message_queue_factory.assert_called_once_with(chain.run_module)
 
 

tests/test_db_subscribe_queries.py

@@ -250,21 +250,6 @@ def test_list_by_type_includes_the_window_start_boundary(db, frozen_now):
     assert "窗口起点上" in async_names and "窗口起点前一秒" not in async_names
 
 
-def test_delete_by_media_identity_removes_matching_seasons_only(db):
-    """
-    按媒体身份删除时，给出季号只删该季，不给则删全部季。
-    """
-    db.add(_sub("第一季", season=1), _sub("第二季", season=2))
-
-    Subscribe().delete_by_media_identity(db.session, TMDB, "9001", season=1)
-
-    remaining = Subscribe.list_by_media_identity(db.session, MediaSource.TMDB, "9001")
-    assert [s.season for s in remaining] == [2]
-
-    Subscribe().delete_by_media_identity(db.session, TMDB, "9001")
-    assert Subscribe.list_by_media_identity(db.session, MediaSource.TMDB, "9001") == []
-
-
 # --------------------------------------------------------------------------- #
 # SubscribeHistory
 # --------------------------------------------------------------------------- #

tests/test_event_contracts.py

@@ -9,8 +9,12 @@ from app.runtime.event.contracts import (
     validate_event_payload,
 )
 from app.runtime.events import Event
+from app.domain.context import Context, MediaInfo
+from app.domain.metainfo import MetaInfo
 from app.schemas.event import ConfigChangeEventData
-from app.schemas.types import ChainEventType, EventType
+from app.schemas.file import FileItem
+from app.schemas.transfer import TransferInfo
+from app.schemas.types import ChainEventType, EventType, MediaType
 
 
 def test_every_event_enum_has_complete_contract() -> None:
@@ -61,6 +65,36 @@ def test_selected_user_side_effects_are_marked_durable_required() -> None:
         assert get_event_contract(event_type).delivery is EventDelivery.DURABLE_REQUIRED
 
 
+def test_download_and_transfer_typed_contracts_accept_legacy_runtime_objects() -> None:
+    """新增 typed contract 只做诊断，不把插件收到的领域对象替换成 dict。"""
+    meta = MetaInfo("Demo.2026.mkv")
+    media = MediaInfo(type=MediaType.MOVIE, title="Demo", year="2026")
+    context = Context(meta_info=meta, media_info=media)
+    fileitem = FileItem(storage="local", path="/downloads/Demo.mkv", type="file")
+    transferinfo = TransferInfo(success=True, fileitem=fileitem)
+
+    assert validate_event_payload(
+        EventType.DownloadAdded,
+        {
+            "hash": "hash-1",
+            "context": context,
+            "downloader": "qb",
+            "episodes": [],
+        },
+    ) == ()
+    for event_type in (EventType.TransferComplete, EventType.TransferFailed):
+        assert validate_event_payload(
+            event_type,
+            {
+                "fileitem": fileitem,
+                "meta": meta,
+                "mediainfo": media,
+                "transferinfo": transferinfo,
+                "transfer_history_id": 1,
+            },
+        ) == ()
+
+
 def test_model_instance_remains_mutable_chain_payload() -> None:
     """链式处理器继续接收原 model 实例，确保输出字段可原地接力。"""
     payload = ConfigChangeEventData(key={"PROXY_HOST"})

tests/test_transfer_overwrite_declined.py

@@ -221,3 +221,53 @@ def test_default_callback_keeps_original_failure_semantics_without_success_histo
         if call.args[0] == EventType.TransferFailed
     ]
     assert len(transfer_failed_events) == 1
+
+
+def test_default_callback_delegates_primary_failure_to_durable_writer():
+    """正式上下文存在 writer 时，主要媒体失败历史和事件必须走同一事务端口。"""
+    chain = make_transfer_chain()
+    chain.eventmanager = MagicMock()
+    chain.post_message = MagicMock()
+    chain.durable_event_writer = MagicMock()
+    task = _make_failed_task()
+    add_fail_calls = []
+    transfer_history_oper = make_history_oper(history=None)
+    transferinfo = TransferInfo(
+        success=False,
+        fileitem=task.fileitem,
+        message="copy failed",
+        transfer_type="copy",
+        need_notify=False,
+    )
+
+    def durable_transfer_result(**kwargs):
+        """执行 writer 收到的历史暂存与提交后发布回调。"""
+        history = kwargs["stage_history"](SimpleNamespace())
+        payload = dict(kwargs["event_payload"])
+        payload["transfer_history_id"] = history.id
+        payload["idempotency_key"] = f"transfer.failed:{history.id}:v1"
+        kwargs["publish"](payload)
+        return history
+
+    chain.durable_event_writer.transfer_result.side_effect = durable_transfer_result
+    with patch(
+        "app.chain.transfer.TransferHistoryOper",
+        return_value=transfer_history_oper,
+    ), patch(
+        "app.chain.transfer.add_transfer_fail",
+        make_fail_recorder(add_fail_calls),
+    ), patch(
+        "app.chain.transfer.settings.AI_AGENT_ENABLE",
+        False,
+    ):
+        state, errmsg = chain._TransferChain__default_callback(task, transferinfo)
+
+    assert state is False
+    assert errmsg == "copy failed"
+    assert len(add_fail_calls) == 1
+    chain.durable_event_writer.transfer_result.assert_called_once()
+    topic = chain.durable_event_writer.transfer_result.call_args.kwargs["topic"]
+    assert topic == "transfer.failed"
+    event_type, event_payload = chain.eventmanager.send_event.call_args.args
+    assert event_type == EventType.TransferFailed
+    assert event_payload["idempotency_key"] == "transfer.failed:1:v1"