refactor(subscribe): make download submission idempotent

app/application/chain/context.py

@@ -12,6 +12,7 @@ from app.application.configuration import ChainRuntimeConfig
 from app.runtime.stop import StopState, runtime_stop_state
 
 if TYPE_CHECKING:
+    from app.application.download.admission import SubscriptionDownloadRepository
     from app.application.download.failures import DownloadFailureRepository
     from app.application.history import (
         DownloadHistoryRepository,
@@ -29,11 +30,11 @@ if TYPE_CHECKING:
         DeleteSubscribeScope,
         SyncDeleteSubscribeScope,
     )
+    from app.application.subscription.execution import SubscriptionSearchRepository
     from app.application.subscription.mutation import (
         SubscriptionMutationScope,
         SyncSubscriptionMutationScope,
     )
-    from app.application.subscription.execution import SubscriptionSearchRepository
     from app.application.transfer.execution import TransferExecutionRepository
     from app.application.transfer.workflow import TransferAdmissionRepository
 
@@ -76,6 +77,7 @@ class ChainRuntimeContext:
     download_failure_repository: DownloadFailureRepository
     user_repository: ChainUserRepository
     subscription_search_repository: Optional[SubscriptionSearchRepository] = None
+    subscription_download_repository: Optional[SubscriptionDownloadRepository] = None
     legacy_transfer_command: Optional[LegacyTransferCommand] = None
     durable_event_writer: Optional[ChainDurableEventWriter] = None
     configuration: ChainRuntimeConfig = field(

app/application/download/admission.py

@@ -0,0 +1,124 @@
+"""订阅下载提交的持久幂等与不确定终态合同。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Optional, Protocol
+
+
+class DownloadReconciliationRequired(RuntimeError):
+    """表示下载器可能已接受任务，必须先对账才能继续自动提交。"""
+
+
+@dataclass(frozen=True, slots=True)
+class SubscriptionDownloadGovernance:
+    """把订阅身份、入口任务与取消检查传入下载提交边界。"""
+
+    subscription_id: int
+    mode: str
+    task_id: Optional[str] = None
+    cancelled: Optional[Callable[[], bool]] = None
+    mark_started: Optional[Callable[[], None]] = None
+
+
+@dataclass(frozen=True, slots=True)
+class SubscriptionDownloadRequest:
+    """一次订阅下载提交认领所需的规范身份。"""
+
+    idempotency_key: str
+    subscription_id: int
+    task_id: Optional[str]
+    logical_identity: str
+    resource_key: str
+    coverage: str
+    mode: str
+
+
+@dataclass(frozen=True, slots=True)
+class SubscriptionDownloadSnapshot:
+    """脱离 Session 的订阅下载提交状态快照。"""
+
+    idempotency_key: str
+    subscription_id: int
+    task_id: Optional[str]
+    state: str
+    attempt_count: int
+    attempt_token: Optional[str]
+    downloader: Optional[str]
+    download_hash: Optional[str]
+    available_at: Optional[str]
+    last_error: Optional[str]
+    created_at: str
+    updated_at: str
+
+
+@dataclass(frozen=True, slots=True)
+class SubscriptionDownloadClaim:
+    """返回本次是否取得唯一提交权以及当前持久状态。"""
+
+    acquired: bool
+    snapshot: SubscriptionDownloadSnapshot
+
+
+class SubscriptionDownloadRepository(Protocol):
+    """订阅下载幂等账本所需的最小持久化端口。"""
+
+    def claim(self, request: SubscriptionDownloadRequest) -> SubscriptionDownloadClaim:
+        """按唯一键认领提交；仅到期 retryable/cancelled 状态允许重新认领。"""
+        ...
+
+    def mark_accepted(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        downloader: Optional[str],
+        download_hash: str,
+    ) -> bool:
+        """记录下载器已明确接受任务，后续任何失败都不得自动重试。"""
+        ...
+
+    def mark_succeeded(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+    ) -> bool:
+        """在 canonical 本地结算完成后写入成功终态。"""
+        ...
+
+    def mark_retryable(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        available_at: str,
+        error: Optional[str],
+    ) -> bool:
+        """记录下载器明确拒绝、尚未产生外部副作用的可重试状态。"""
+        ...
+
+    def mark_reconcile_required(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        error: Optional[str],
+        downloader: Optional[str] = None,
+        download_hash: Optional[str] = None,
+    ) -> bool:
+        """冻结可能已产生外部副作用的提交，等待下载器对账。"""
+        ...
+
+    def mark_cancelled(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+    ) -> bool:
+        """仅在进入下载器副作用前收口取消。"""
+        ...
+
+    def has_started_for_task(self, task_id: str) -> bool:
+        """判断搜索任务是否已有不能按未执行处理的下载提交。"""
+        ...

app/application/subscription/candidates.py

@@ -3,15 +3,14 @@
 import copy
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional, cast
 from uuid import uuid4
 
 from app.application.subscription.contract import SubscriptionSnapshot
 from app.domain.context import Context, MediaInfo
 from app.foundation import text as text_tools
 from app.schemas.media import resolve_media_identity
-from app.schemas.types import MediaType
-
+from app.schemas.types import MediaSource, MediaType
 
 CandidateGroups = Dict[str, List[Context]]
 
@@ -196,11 +195,11 @@ class CandidateIndex:
         context.candidate_recognized = False
         context.media_info_is_target = True
         context.media_info = MediaInfo(
-            type=subscribe.type,
+            type=cast(MediaType, subscribe.type),
             title=subscribe.name,
-            media_source=subscribe.media_source,
-            media_id=subscribe.media_id,
-            season=subscribe.season,
+            media_source=cast(MediaSource, subscribe.media_source),
+            media_id=cast(str, subscribe.media_id),
+            season=cast(int, subscribe.season),
         )
 
     @classmethod
@@ -236,7 +235,7 @@ class CandidateIndex:
         return media_season is None or target_season == media_season
 
     @classmethod
-    def meta_seasons(cls, meta_info) -> set[int]:
+    def meta_seasons(cls, meta_info: Any) -> set[int]:
         """提取标题解析出的显式季范围。"""
         meta_fields = vars(meta_info) if meta_info else {}
         if "season_list" in meta_fields:
@@ -271,7 +270,7 @@ class CandidateIndex:
         return {identity for identity in identities if identity}
 
     @staticmethod
-    def media_identity(media) -> Optional[tuple[str, str]]:
+    def media_identity(media: Any) -> Optional[tuple[str, str]]:
         """把动态媒体对象的身份归一为可索引键。"""
         source, media_id = resolve_media_identity(media=media)
         if not source or not media_id:
@@ -279,7 +278,7 @@ class CandidateIndex:
         return str(source), media_id
 
     @staticmethod
-    def normalize_int(value) -> Optional[int]:
+    def normalize_int(value: Any) -> Optional[int]:
         """将季号等动态字段转为整数。"""
         if value is None:
             return None
@@ -289,15 +288,15 @@ class CandidateIndex:
             return None
 
     @staticmethod
-    def normalize_media_type(value) -> Optional[str]:
+    def normalize_media_type(value: Any) -> Optional[str]:
         """统一媒体类型枚举与字符串形态。"""
         if isinstance(value, MediaType):
             value = value.value
         if value == MediaType.UNKNOWN.value:
             return None
-        return value
+        return str(value) if value is not None else None
 
     @staticmethod
-    def normalize_title(value) -> str:
+    def normalize_title(value: Any) -> str:
         """归一标题用于低置信标题匹配。"""
         return (text_tools.normalize_upper(value or "") or "").strip()

app/application/subscription/sitebudget.py

@@ -148,7 +148,7 @@ class SubscriptionSiteBudget:
             "login_invalid": (900.0, 21600.0),
             "timeout": (300.0, 7200.0),
         }.get(outcome, (180.0, 3600.0))
-        return min(base * (2**exponent), ceiling)
+        return float(min(base * (2**exponent), ceiling))
 
     def _raise_if_cancelled(self) -> None:
         """在不持有业务锁的等待边界传播取消或停机。"""

app/chain/base.py

@@ -72,6 +72,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
         self.transfer_execution_repository = context.transfer_execution_repository
         self.media_server_repository = context.media_server_repository
         self.download_failure_repository = context.download_failure_repository
+        self.subscription_download_repository = context.subscription_download_repository
         self.user_repository = context.user_repository
         self.runtime_config = context.configuration
         self.stop_state = context.stop_state

app/chain/download/admission.py

@@ -0,0 +1,201 @@
+"""订阅下载提交幂等身份与状态转换 owner。"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timedelta, timezone
+from typing import Optional, Set, cast
+
+from app.application.download.admission import (
+    DownloadReconciliationRequired,
+    SubscriptionDownloadClaim,
+    SubscriptionDownloadGovernance,
+    SubscriptionDownloadRepository,
+    SubscriptionDownloadRequest,
+)
+from app.chain.download.contract import _DownloadOwnerBase
+from app.domain import episode as episode_rules
+from app.domain.context import Context
+from app.schemas.media import build_media_key, resolve_media_identity
+from app.schemas.types import MediaType
+
+
+class DownloadAdmissionOwner(_DownloadOwnerBase):
+    """计算订阅提交唯一键并通过持久账本控制外部副作用。"""
+
+    def _build_subscription_download_request(
+        self,
+        *,
+        context: Context,
+        episodes: Optional[Set[int]],
+        governance: SubscriptionDownloadGovernance,
+    ) -> SubscriptionDownloadRequest:
+        """组合订阅、torrent、季集覆盖与模式生成规范幂等请求。"""
+        media = context.media_info
+        meta = context.meta_info
+        torrent = context.torrent_info
+        media_source, media_id = resolve_media_identity(media=media)
+        media_key = build_media_key(media_source, media_id) or (
+            f"{getattr(media, 'title', '')}:{getattr(media, 'year', '')}"
+        )
+        seasons = sorted(set(getattr(meta, "season_list", None) or []))
+        selected = sorted(set(episodes or getattr(meta, "episode_list", None) or []))
+        if selected:
+            coverage = f"episodes:{episode_rules.format_ranges(selected)}"
+        elif seasons:
+            coverage = "seasons:" + ",".join(str(season) for season in seasons) + ":full"
+        else:
+            coverage = "full"
+        media_type = getattr(media, "type", None)
+        media_type_value = getattr(media_type, "value", media_type)
+        media_season = getattr(media, "season", None)
+        meta_season = getattr(meta, "season", None)
+        logical_identity = json.dumps(
+            {
+                "subscription_id": governance.subscription_id,
+                "media_key": str(media_key or ""),
+                "media_type": str(media_type_value or ""),
+                "season": media_season if media_season is not None else meta_season,
+                "episode_group": getattr(media, "episode_group", None),
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+        resource_key = self._torrent_resource_key(torrent)
+        canonical = json.dumps(
+            {
+                "logical_identity": logical_identity,
+                "resource_key": resource_key,
+                "coverage": coverage,
+                "mode": governance.mode,
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+        return SubscriptionDownloadRequest(
+            idempotency_key=hashlib.sha256(canonical.encode("utf-8")).hexdigest(),
+            subscription_id=governance.subscription_id,
+            task_id=governance.task_id,
+            logical_identity=logical_identity,
+            resource_key=resource_key,
+            coverage=coverage,
+            mode=governance.mode,
+        )
+
+    def _claim_subscription_download(
+        self,
+        *,
+        context: Context,
+        episodes: Optional[Set[int]],
+        governance: Optional[SubscriptionDownloadGovernance],
+    ) -> tuple[Optional[SubscriptionDownloadClaim], Optional[str]]:
+        """在下载器调用前认领唯一提交权，并返回已成功提交的历史 hash。"""
+        if governance is None:
+            return None, None
+        legacy_hash = self._legacy_subscription_download_hash(
+            context=context,
+            episodes=episodes,
+            governance=governance,
+        )
+        if legacy_hash:
+            return None, legacy_hash
+        repository = getattr(self, "subscription_download_repository", None)
+        if repository is None:
+            raise RuntimeError("订阅下载幂等仓储尚未配置")
+        request = self._build_subscription_download_request(
+            context=context,
+            episodes=episodes,
+            governance=governance,
+        )
+        claim = repository.claim(request)
+        snapshot = claim.snapshot
+        if claim.acquired:
+            if not snapshot.attempt_token:
+                raise RuntimeError("订阅下载提交已认领但缺少尝试令牌")
+            return claim, None
+        if snapshot.state == "succeeded" and snapshot.download_hash:
+            return claim, snapshot.download_hash
+        if snapshot.state in {"submitting", "accepted", "reconcile_required"}:
+            raise DownloadReconciliationRequired(
+                f"订阅下载提交 {snapshot.idempotency_key} 当前为 {snapshot.state}，需要先对账下载器"
+            )
+        return claim, None
+
+    def _legacy_subscription_download_hash(
+        self,
+        *,
+        context: Context,
+        episodes: Optional[Set[int]],
+        governance: SubscriptionDownloadGovernance,
+    ) -> Optional[str]:
+        """重读迁移前下载历史，兼容识别同订阅同 torrent 同覆盖的成功提交。"""
+        media = context.media_info
+        meta = context.meta_info
+        torrent = context.torrent_info
+        repository = getattr(self, "download_history_repository", None)
+        media_source, media_id = resolve_media_identity(media=media)
+        if repository is None or not media_source or not media_id or not torrent:
+            return None
+        histories = repository.get_by_media_identity(
+            media_source=media_source,
+            media_id=str(media_id),
+            music_type=getattr(media, "music_type", None),
+        )
+        expected_episodes = episode_rules.format_ranges(
+            sorted(set(episodes or getattr(meta, "episode_list", None) or []))
+        )
+        expected_season = str(getattr(meta, "season", None) or "")
+        for history in histories:
+            if not history.download_hash:
+                continue
+            if history.torrent_name != torrent.title or history.torrent_site != torrent.site_name:
+                continue
+            if not self._history_matches_subscription(history.note, governance.subscription_id):
+                continue
+            if getattr(history, "episode_group", None) != getattr(media, "episode_group", None):
+                continue
+            if expected_episodes:
+                if history.episodes == expected_episodes:
+                    return cast(str, history.download_hash)
+                continue
+            if getattr(media, "type", None) == MediaType.TV and str(history.seasons or "") != expected_season:
+                continue
+            return cast(str, history.download_hash)
+        return None
+
+    @staticmethod
+    def _history_matches_subscription(note: object, subscription_id: int) -> bool:
+        """从下载历史来源快照确认记录属于同一订阅，而非同媒体其他记录。"""
+        if not isinstance(note, dict):
+            return False
+        source = note.get("source")
+        if not isinstance(source, str) or not source.startswith("Subscribe|"):
+            return False
+        try:
+            payload = json.loads(source.split("|", 1)[1])
+            return int(payload.get("id")) == subscription_id
+        except (AttributeError, TypeError, ValueError, json.JSONDecodeError):
+            return False
+
+    @staticmethod
+    def _subscription_download_cancelled(
+        governance: Optional[SubscriptionDownloadGovernance],
+    ) -> bool:
+        """在可安全取消边界调用入口提供的取消检查。"""
+        return bool(governance and governance.cancelled and governance.cancelled())
+
+    @staticmethod
+    def _subscription_download_retry_at(error: Optional[str], ttl_seconds: int) -> str:
+        """把明确拒绝的下载提交推迟到失败冷却到期后。"""
+        del error
+        return (
+            datetime.now(timezone.utc) + timedelta(seconds=max(1, ttl_seconds))
+        ).isoformat(timespec="seconds")
+
+    def _subscription_download_repository(self) -> SubscriptionDownloadRepository:
+        """返回已配置的订阅下载仓储，缺失时拒绝越过外部副作用边界。"""
+        repository = getattr(self, "subscription_download_repository", None)
+        if repository is None:
+            raise RuntimeError("订阅下载幂等仓储尚未配置")
+        return cast(SubscriptionDownloadRepository, repository)

app/chain/download/batch.py

@@ -4,6 +4,7 @@ import copy
 from typing import Callable, Dict, List, Optional, Set, Tuple, cast
 
 from app.application.download import selection as _selection
+from app.application.download.admission import SubscriptionDownloadGovernance
 from app.application.torrent.download import TorrentHelper
 from app.chain.download.contract import _DownloadOwnerBase
 from app.domain import episode as episode_rules
@@ -39,7 +40,8 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                        userid: Optional[str] = None,
                        username: Optional[str] = None,
                        downloader: Optional[str] = None,
-                       custom_words: Optional[str] = None
+                       custom_words: Optional[str] = None,
+                       governance: Optional[SubscriptionDownloadGovernance] = None,
                        ) -> Tuple[
                            List[Context],
                            Optional[Dict[str, Dict[int, NotExistMediaInfo]]],
@@ -60,6 +62,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
             username=username,
             downloader=downloader,
             custom_words=custom_words,
+            governance=governance,
         )
 
     def _execute_batch_download(self,
@@ -71,7 +74,8 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                        userid: Optional[str] = None,
                        username: Optional[str] = None,
                        downloader: Optional[str] = None,
-                       custom_words: Optional[str] = None
+                       custom_words: Optional[str] = None,
+                       governance: Optional[SubscriptionDownloadGovernance] = None,
                        ) -> Tuple[
                            List[Context],
                            Optional[Dict[str, Dict[int, NotExistMediaInfo]]],
@@ -87,6 +91,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
         :param username: 调用下载的用户名/插件名
         :param downloader: 下载器
         :param custom_words: 下载来源（如订阅）的完整自定义识别词文本，随下载记录存档，供整理时原样复现识别
+        :param governance: 订阅下载幂等、入口任务和取消边界；非订阅调用保持为空
         :return: 已下载资源列表及剩余缺集，键格式为 no_exists[source:id]
         """
         no_exists_was_none = no_exists is None
@@ -186,6 +191,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
             username=username,
             downloader=downloader,
             custom_words=custom_words,
+            governance=governance,
         )
 
         # 电视剧整季匹配
@@ -303,7 +309,8 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                         userid=userid,
                                         username=username,
                                         downloader=downloader,
-                                        custom_words=custom_words
+                                        custom_words=custom_words,
+                                        governance=governance,
                                     )
                             else:
                                 # 下载
@@ -312,7 +319,8 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                                                    channel=channel, source=source,
                                                                    userid=userid, username=username,
                                                                    downloader=downloader,
-                                                                   custom_words=custom_words)
+                                                                   custom_words=custom_words,
+                                                                   governance=governance)
 
                             if download_id:
                                 # 下载成功
@@ -401,7 +409,8 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                                                    channel=channel, source=source,
                                                                    userid=userid, username=username,
                                                                    downloader=downloader,
-                                                                   custom_words=custom_words)
+                                                                   custom_words=custom_words,
+                                                                   governance=governance)
                                 if download_id:
                                     # 下载成功
                                     if __requires_complete_coverage(missing_info):
@@ -518,7 +527,8 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                 userid=userid,
                                 username=username,
                                 downloader=downloader,
-                                custom_words=custom_words
+                                custom_words=custom_words,
+                                governance=governance,
                             )
                             if not download_id:
                                 __remember_context_failure(context)

app/chain/download/contract.py

@@ -9,6 +9,7 @@ if TYPE_CHECKING:
         """向类型检查器声明同一 Facade 上的跨 owner 方法。"""
 
         _active_download_failure_fingerprints: Callable[..., Any]
+        _build_subscription_download_request: Callable[..., Any]
         _append_no_exists: Callable[..., None]
         _build_download_notification: Callable[..., Any]
         _build_download_failure_fingerprint: Callable[..., Any]
@@ -22,14 +23,20 @@ if TYPE_CHECKING:
         _is_job_active: Callable[..., bool]
         _is_subscribe_source: Callable[..., bool]
         _log_download_failure_cooldown: Callable[..., None]
+        _legacy_subscription_download_hash: Callable[..., Any]
+        _history_matches_subscription: Callable[..., bool]
         _media_identity_keys: Callable[..., set[str]]
         _matches_media_identity: Callable[..., bool]
         _prepare_batch_download_contexts: Callable[..., Any]
+        _claim_subscription_download: Callable[..., Any]
         _record_download_failure: Callable[..., Any]
         _resolve_media_download_dir: Callable[..., Any]
         _settle_download_success: Callable[..., None]
         _submit_download_added_task: Callable[..., None]
         _torrent_resource_key: Callable[..., str]
+        _subscription_download_cancelled: Callable[..., bool]
+        _subscription_download_repository: Callable[..., Any]
+        _subscription_download_retry_at: Callable[..., str]
         _validate_music_album_resource: Callable[..., Any]
         batch_download: Callable[..., Any]
         download_single: Callable[..., Any]

app/chain/download/facade.py

@@ -3,6 +3,7 @@
 from collections.abc import Callable
 from typing import Any, TypeVar
 
+from app.chain.download.admission import DownloadAdmissionOwner
 from app.chain.download.batch import DownloadBatchOwner
 from app.chain.download.existence import DownloadExistenceOwner
 from app.chain.download.failure import DownloadFailureOwner
@@ -29,6 +30,7 @@ class DownloadChain(
     DownloadSubtitleOwner,
     DownloadSelectionOwner,
     DownloadFailureOwner,
+    DownloadAdmissionOwner,
     DownloadSubmissionOwner,
     DownloadBatchOwner,
     DownloadExistenceOwner,

app/chain/download/selection.py

@@ -4,6 +4,7 @@ from pathlib import Path
 from typing import Callable, Dict, List, Optional, Set, Tuple, Union, cast
 
 from app.application.configuration import get_chain_runtime_config_snapshot
+from app.application.download.admission import SubscriptionDownloadGovernance
 from app.application.download.failures import (
     DownloadFailureSnapshot,
 )
@@ -156,6 +157,7 @@ class DownloadSelectionOwner(_DownloadOwnerBase):
         username: Optional[str],
         downloader: Optional[str],
         custom_words: Optional[str],
+        governance: Optional[SubscriptionDownloadGovernance],
     ) -> None:
         """
         处理电影与音乐的直接候选下载。
@@ -201,6 +203,7 @@ class DownloadSelectionOwner(_DownloadOwnerBase):
                 username=username,
                 downloader=downloader,
                 custom_words=custom_words,
+                governance=governance,
             ):
                 logger.info(f"{context.torrent_info.title} 添加下载成功")
                 downloaded_list.append(context)

app/chain/download/submission.py

@@ -9,6 +9,10 @@ from urllib.parse import urlencode, urljoin, urlparse
 
 from app.application.configuration import get_chain_runtime_config_snapshot
 from app.application.directory import validate_download_save_path
+from app.application.download.admission import (
+    DownloadReconciliationRequired,
+    SubscriptionDownloadGovernance,
+)
 from app.application.torrent.download import TorrentHelper
 from app.chain.download.contract import _DownloadOwnerBase
 from app.chain.download.ports import (
@@ -270,7 +274,9 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
                         username: Optional[str] = None,
                         label: Optional[str] = None,
                         return_detail: bool = False,
-                        custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
+                        custom_words: Optional[str] = None,
+                        governance: Optional[SubscriptionDownloadGovernance] = None,
+                        ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
         """
         下载单个资源并发送结果通知。
 
@@ -290,6 +296,7 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
             label=label,
             return_detail=return_detail,
             custom_words=custom_words,
+            governance=governance,
         )
 
     def _execute_download_single(self, context: Context,
@@ -304,7 +311,9 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
                         username: Optional[str] = None,
                         label: Optional[str] = None,
                         return_detail: bool = False,
-                        custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
+                        custom_words: Optional[str] = None,
+                        governance: Optional[SubscriptionDownloadGovernance] = None,
+                        ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
         """
         下载及发送通知
         :param context: 资源上下文
@@ -320,6 +329,7 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
         :param label: 自定义标签
         :param return_detail: 是否返回详细结果；False 时返回下载任务 hash 或 None，True 时返回 (hash, error_msg)
         :param custom_words: 下载来源（如订阅）的完整自定义识别词文本，随下载记录存档，供整理时原样复现识别
+        :param governance: 订阅级提交幂等、入口任务与取消检查；非订阅调用保持为空
         :return: return_detail=False 时返回下载任务 hash 或 None；return_detail=True 时返回 (hash, error_msg)
         """
         _torrent = context.torrent_info
@@ -408,20 +418,77 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
         file_uri = FileURI(storage=storage, path=download_dir.as_posix())
         download_dir = Path(file_uri.uri)
 
+        if self._subscription_download_cancelled(governance):
+            cancel_error = "订阅下载在提交前已取消"
+            return (None, cancel_error) if return_detail else None
+        admission, duplicate_hash = self._claim_subscription_download(
+            context=context,
+            episodes=episodes,
+            governance=governance,
+        )
+        if duplicate_hash:
+            if governance and governance.mark_started:
+                governance.mark_started()
+            logger.info(f"{_torrent.title} 已由重叠订阅入口提交，复用任务 {duplicate_hash}")
+            return (duplicate_hash, "下载任务已由重叠入口提交") if return_detail else duplicate_hash
+        if admission is not None and not admission.acquired:
+            wait_error = (
+                f"订阅下载提交当前为 {admission.snapshot.state}，"
+                f"最早可重试：{admission.snapshot.available_at or '待下一轮'}"
+            )
+            return (None, wait_error) if return_detail else None
+        attempt_token = admission.snapshot.attempt_token if admission is not None else None
+        admission_key = admission.snapshot.idempotency_key if admission is not None else None
+        if admission is not None and self._subscription_download_cancelled(governance):
+            self._subscription_download_repository().mark_cancelled(
+                idempotency_key=admission.snapshot.idempotency_key,
+                attempt_token=admission.snapshot.attempt_token or "",
+            )
+            cancel_error = "订阅下载在下载器调用前已取消"
+            return (None, cancel_error) if return_detail else None
+
         # 添加下载
-        result: Optional[Tuple[Optional[str], Optional[str], Optional[str], str]] = self.download(content=torrent_content,
+        if governance and governance.mark_started:
+            governance.mark_started()
+        try:
+            result: Optional[Tuple[Optional[str], Optional[str], Optional[str], str]] = self.download(
+                content=torrent_content,
                 cookie=_torrent.site_cookie,
                 episodes=cast(Set[int], episodes),
                 download_dir=download_dir,
                 category=_media.category,
                 label=label,
-                                                downloader=downloader or _site_downloader)
+                downloader=downloader or _site_downloader,
+            )
+        except Exception as err:
+            if admission_key and attempt_token:
+                self._subscription_download_repository().mark_reconcile_required(
+                    idempotency_key=admission_key,
+                    attempt_token=attempt_token,
+                    error=f"下载器调用异常：{str(err)}",
+                )
+                raise DownloadReconciliationRequired(
+                    f"{_torrent.title} 下载器结果不确定，已冻结自动重试"
+                ) from err
+            raise
         if result:
             _downloader, _hash, _layout, error_msg = result
         else:
             _downloader, _hash, _layout, error_msg = None, None, None, "未找到下载器"
 
         if _hash:
+            if admission_key and attempt_token:
+                accepted = self._subscription_download_repository().mark_accepted(
+                    idempotency_key=admission_key,
+                    attempt_token=attempt_token,
+                    downloader=_downloader,
+                    download_hash=_hash,
+                )
+                if not accepted:
+                    raise DownloadReconciliationRequired(
+                        f"{_torrent.title} 已被下载器接受，但本地接受状态写入失败"
+                    )
+            try:
                 self._settle_download_success(
                     context=context,
                     media=_media,
@@ -442,7 +509,47 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
                     torrent_content=torrent_content,
                     custom_words=custom_words,
                 )
+            except Exception as err:
+                if admission_key and attempt_token:
+                    self._subscription_download_repository().mark_reconcile_required(
+                        idempotency_key=admission_key,
+                        attempt_token=attempt_token,
+                        error=f"下载器已接受但本地结算失败：{str(err)}",
+                        downloader=_downloader,
+                        download_hash=_hash,
+                    )
+                    raise DownloadReconciliationRequired(
+                        f"{_torrent.title} 已被下载器接受但本地结算失败，已转待对账"
+                    ) from err
+                raise
+            if admission_key and attempt_token:
+                succeeded = self._subscription_download_repository().mark_succeeded(
+                    idempotency_key=admission_key,
+                    attempt_token=attempt_token,
+                )
+                if not succeeded:
+                    self._subscription_download_repository().mark_reconcile_required(
+                        idempotency_key=admission_key,
+                        attempt_token=attempt_token,
+                        error="本地结算完成但幂等成功终态写入失败",
+                        downloader=_downloader,
+                        download_hash=_hash,
+                    )
+                    raise DownloadReconciliationRequired(
+                        f"{_torrent.title} 本地结算完成但提交终态未确认，已转待对账"
+                    )
         else:
+            if admission_key and attempt_token:
+                retry_at = self._subscription_download_retry_at(
+                    error_msg,
+                    self._download_failure_ttl(error_msg),
+                )
+                self._subscription_download_repository().mark_retryable(
+                    idempotency_key=admission_key,
+                    attempt_token=attempt_token,
+                    available_at=retry_at,
+                    error=error_msg,
+                )
             # 下载失败
             logger.error(f"{_media.title_year} 添加下载任务失败："
                          f"{_torrent.title} - {_torrent.enclosure}，{error_msg}")

app/chain/search/facade.py

@@ -4,6 +4,7 @@ import threading
 from collections.abc import AsyncIterator, Callable
 from typing import Any, Optional, TypeVar, cast
 
+from app.application.subscription.sitebudget import SubscriptionSiteBudget
 from app.chain.base import ChainBase
 from app.chain.search.cache import SearchCacheOwner
 from app.chain.search.media import SearchMediaOwner
@@ -16,7 +17,6 @@ from app.chain.search.result import SearchResultOwner
 from app.chain.search.site import SearchSiteOwner
 from app.chain.search.subtitle import SearchSubtitleOwner
 from app.chain.search.title import SearchTitleOwner
-from app.application.subscription.sitebudget import SubscriptionSiteBudget
 from app.domain.context import Context, MediaInfo, SubtitleInfo
 from app.runtime.events import Event, eventmanager
 from app.schemas.mediaserver import NotExistMediaInfo

app/chain/search/result.py

@@ -203,8 +203,7 @@ def _match_torrents(
 def _context_media(mediainfo: MediaInfo) -> MediaInfo:
     """复制并裁剪上下文媒体信息，避免修改调用方持有的目标对象。"""
     context_media = copy.copy(mediainfo)
-    clear_context_media = cast(Callable[[], None], context_media.clear)
-    clear_context_media()
+    context_media.clear()
     return context_media
 
 

app/chain/subscribe/completion.py

@@ -189,7 +189,7 @@ class SubscribeCompletionOwner(_SubscribeOwnerBase):
             mediainfo=mediainfo,
             mediakey=mediakey,
         )
-        return completed
+        return bool(completed)
 
     def _SubscribeChain__update_subscribe_note(
         self,

app/chain/subscribe/policy.py

@@ -1,8 +1,9 @@
 """订阅优先级、剧集范围与来源编码策略"""
 
 import json
-from typing import Any, Dict, List, Optional, Tuple, Union, cast
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast
 
+from app.application.download.admission import SubscriptionDownloadGovernance
 from app.application.subscription import priority as _priority
 from app.application.subscription.contract import (
     SubscriptionSnapshot,
@@ -258,6 +259,14 @@ class SubscribePolicyOwner(_SubscribeOwnerBase):
         )
         return result
 
+    @staticmethod
+    def _SubscribeChain__load_current_subscription(
+        loader: Callable[[int], Optional[SubscriptionSnapshot]],
+        subscription_id: int,
+    ) -> Optional[SubscriptionSnapshot]:
+        """通过显式可调用边界读取提交前的最新订阅快照。"""
+        return loader(subscription_id)
+
     def _SubscribeChain__download_best_version_with_full_pack_first(
         self,
         contexts: List[Context],
@@ -272,6 +281,60 @@ class SubscribePolicyOwner(_SubscribeOwnerBase):
         """
         TV 分集洗版先尝试覆盖目标范围的全集资源，失败后回退到按集下载。
         """
+        governance: Optional[SubscriptionDownloadGovernance] = None
+        repository = getattr(self, "subscription_repository", None)
+        get_current = getattr(repository, "get", None)
+        if callable(get_current):
+            current_get = cast(
+                Callable[[int], Optional[SubscriptionSnapshot]],
+                get_current,
+            )
+            current = self._SubscribeChain__load_current_subscription(
+                current_get,
+                subscribe.id,
+            )
+            if current is None:
+                logger.info(f"订阅 {subscribe.id} 已删除，放弃本轮下载提交")
+                return [], no_exists
+            if current.state == "S":
+                logger.info(f"订阅 {current.name} 已暂停，放弃本轮下载提交")
+                return [], no_exists
+            if self._SubscribeChain__candidate_contract_changed(subscribe, current):
+                logger.info(f"订阅 {current.name} 的筛选或媒体身份已变化，放弃旧候选并等待下一轮")
+                return [], no_exists
+            if not contexts or not contexts[0].meta_info or not contexts[0].media_info:
+                return [], no_exists
+
+            exists, fresh_no_exists = self.check_and_handle_existing_media(
+                subscribe=current,
+                meta=contexts[0].meta_info,
+                mediainfo=contexts[0].media_info,
+                mediakey=mediakey,
+            )
+            if exists:
+                return [], fresh_no_exists
+            contexts = self._SubscribeChain__revalidate_download_contexts(current, contexts)
+            if not contexts:
+                return [], fresh_no_exists
+            subscribe = current
+            no_exists = fresh_no_exists
+            username = current.username
+            save_path = current.save_path
+            downloader = current.downloader
+            source = self.get_subscribe_source_keyword(current)
+            governance = SubscriptionDownloadGovernance(
+                subscription_id=current.id,
+                mode=self._SubscribeChain__download_governance_mode(current),
+                task_id=cast(Optional[str], getattr(self, "_subscription_download_task_id", None)),
+                cancelled=cast(
+                    Optional[Callable[[], bool]],
+                    getattr(self, "_subscription_download_cancelled", None),
+                ),
+                mark_started=cast(
+                    Optional[Callable[[], None]],
+                    getattr(self, "_subscription_download_mark_started", None),
+                ),
+            )
         full_pack_no_exists = self._SubscribeChain__build_full_pack_first_no_exists(
             subscribe=subscribe, mediakey=mediakey
         )
@@ -316,6 +379,7 @@ class SubscribePolicyOwner(_SubscribeOwnerBase):
                 downloader=downloader,
                 source=source,
                 custom_words=subscribe.custom_words,
+                governance=governance,
             )
             if downloads:
                 return downloads, lefts
@@ -330,10 +394,84 @@ class SubscribePolicyOwner(_SubscribeOwnerBase):
                 downloader=downloader,
                 source=source,
                 custom_words=subscribe.custom_words,
+                governance=governance,
             )
         )
         return result
 
+    @staticmethod
+    def _SubscribeChain__candidate_contract_changed(
+        prepared: SubscriptionSnapshot,
+        current: SubscriptionSnapshot,
+    ) -> bool:
+        """检测会让准备阶段候选失效的订阅身份和筛选字段变化。"""
+        fields = (
+            "type",
+            "media_source",
+            "media_id",
+            "music_type",
+            "season",
+            "episode_group",
+            "keyword",
+            "sites",
+            "include",
+            "exclude",
+            "quality",
+            "resolution",
+            "effect",
+            "audio_quality",
+            "audio_format",
+            "min_bitrate",
+            "min_bit_depth",
+            "min_sample_rate",
+            "filter_groups",
+            "custom_words",
+        )
+        return any(getattr(prepared, field) != getattr(current, field) for field in fields)
+
+    def _SubscribeChain__revalidate_download_contexts(
+        self,
+        subscribe: SubscriptionSnapshot,
+        contexts: List[Context],
+    ) -> List[Context]:
+        """按当前洗版模式和优先级重新过滤准备阶段候选。"""
+        if not subscribe.best_version:
+            return contexts
+        accepted: List[Context] = []
+        for context in contexts:
+            media = context.media_info
+            meta = context.meta_info
+            torrent = context.torrent_info
+            if not media or not meta or not torrent:
+                continue
+            if media.type == MediaType.TV:
+                if self._SubscribeChain__is_full_best_version_enabled(subscribe) \
+                        and not self._SubscribeChain__is_full_season_resource(meta, subscribe):
+                    continue
+                if not self._is_episode_range_covered(meta, subscribe):
+                    continue
+                if not self._SubscribeChain__prepare_best_version_tv_candidate(
+                    subscribe=subscribe,
+                    context=context,
+                    priority=torrent.pri_order,
+                ):
+                    continue
+            elif subscribe.current_priority and torrent.pri_order <= subscribe.current_priority:
+                continue
+            accepted.append(context)
+        return accepted
+
+    @staticmethod
+    def _SubscribeChain__download_governance_mode(subscribe: SubscriptionSnapshot) -> str:
+        """把订阅模式归一为稳定幂等键组成部分。"""
+        if not subscribe.best_version:
+            return "normal"
+        if subscribe.type == MediaType.TV.value and subscribe.best_version_full:
+            return "best_version_full"
+        if subscribe.type == MediaType.TV.value:
+            return "best_version_episode"
+        return "best_version"
+
     @classmethod
     def _is_episode_range_covered(cls, meta: MetaBase, subscribe: SubscriptionSnapshot) -> bool:
         """

app/chain/subscribe/search.py

@@ -3,6 +3,7 @@
 import random
 import time
 from datetime import datetime, timedelta, timezone
+from functools import partial
 from typing import Any, Callable, Optional, cast
 from uuid import uuid4
 
@@ -13,8 +14,8 @@ from app.application.subscription.contract import (
     build_subscribe_meta,
     subscribe_media_key,
 )
-from app.application.subscription.query import SubscriptionQueryService
 from app.application.subscription.execution import SearchBatchSnapshot, SubscriptionSearchRepository
+from app.application.subscription.query import SubscriptionQueryService
 from app.application.subscription.sitebudget import (
     SubscriptionSearchCancelled,
     SubscriptionSiteBudget,
@@ -118,9 +119,8 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
         :param sids: 订阅ID集合，有值时按给定顺序处理
         :return: 更新订阅状态为R或删除订阅
         """
-        queue = cast(
-            Optional[SubscriptionSearchRepository],
-            getattr(self, "subscription_search_repository", None),
+        queue: Optional[SubscriptionSearchRepository] = getattr(
+            self, "subscription_search_repository", None
         )
         if queue is not None:
             return self._execute_queued_search(
@@ -246,7 +246,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                     "coalesced": enqueued.coalesced_count,
                 },
             )
-        return enqueued.batch.batch_id
+        return str(enqueued.batch.batch_id)
 
     def _drain_search_queue(
         self,
@@ -273,6 +273,8 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                 if not task.lease_token:
                     logger.error(f"订阅搜索任务 {task.task_id} 缺少租约令牌，跳过执行")
                     continue
+                task_id = str(task.task_id)
+                cancelled = partial(queue.is_cancel_requested, task_id)
                 if queue.is_cancel_requested(task.task_id):
                     queue.release_task(
                         task_id=task.task_id,
@@ -301,14 +303,26 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                 searchchain.configure_subscription_site_budget(
                     SubscriptionSiteBudget(
                         repository=queue,
-                        owner=f"{owner}:{task.task_id}",
-                        cancelled=lambda task_id=task.task_id: queue.is_cancel_requested(task_id),
+                        owner=f"{owner}:{task_id}",
+                        cancelled=cancelled,
                         stop_state=getattr(self, "stop_state", runtime_stop_state),
                     )
                 )
+                self._subscription_download_task_id = task_id
+                self._subscription_download_cancelled = cancelled
+                self._subscription_download_crossed_boundary = False
+                self._subscription_download_mark_started = self._mark_subscription_download_started
                 try:
                     current = self._process_search_subscription(subscribe, searchchain)
                     if queue.is_cancel_requested(task.task_id):
+                        if self._subscription_download_started_for_task(task.task_id):
+                            queue.finish_task(
+                                task_id=task.task_id,
+                                lease_token=task.lease_token,
+                                state="completed",
+                                error="取消请求晚于下载提交边界，已按实际结果完成",
+                            )
+                        else:
                             queue.release_task(
                                 task_id=task.task_id,
                                 lease_token=task.lease_token,
@@ -336,6 +350,10 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                         error=str(err),
                     )
                 finally:
+                    delattr(self, "_subscription_download_task_id")
+                    delattr(self, "_subscription_download_cancelled")
+                    delattr(self, "_subscription_download_mark_started")
+                    self._subscription_download_crossed_boundary = False
                     searchchain.configure_subscription_site_budget(None)
                     if current and current.state == "N":
                         try:
@@ -355,15 +373,25 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
                 queue_lock.release()
         return processed
 
+    def _subscription_download_started_for_task(self, task_id: str) -> bool:
+        """判断取消是否已晚于下载器副作用边界。"""
+        if bool(getattr(self, "_subscription_download_crossed_boundary", False)):
+            return True
+        repository = getattr(self, "subscription_download_repository", None)
+        return bool(repository and repository.has_started_for_task(task_id))
+
+    def _mark_subscription_download_started(self) -> None:
+        """记录当前搜索任务已提交或复用了真实下载结果。"""
+        self._subscription_download_crossed_boundary = True
+
     def resume_search_queue(
         self,
         progress_callback: Optional[Callable[..., None]] = None,
         limit: int = 50,
     ) -> None:
         """短周期恢复排队或租约过期任务，不创建新的 24 小时兜底批次。"""
-        queue = cast(
-            Optional[SubscriptionSearchRepository],
-            getattr(self, "subscription_search_repository", None),
+        queue: Optional[SubscriptionSearchRepository] = getattr(
+            self, "subscription_search_repository", None
         )
         if queue is None:
             return
@@ -413,17 +441,15 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
 
     def cancel_search_batch(self, batch_id: str) -> bool:
         """请求取消持久搜索批次；未注入队列时返回失败。"""
-        queue = cast(
-            Optional[SubscriptionSearchRepository],
-            getattr(self, "subscription_search_repository", None),
+        queue: Optional[SubscriptionSearchRepository] = getattr(
+            self, "subscription_search_repository", None
         )
         return bool(queue and queue.request_cancel(batch_id))
 
     def get_search_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]:
         """返回持久搜索批次状态；未注入队列时返回空。"""
-        queue = cast(
-            Optional[SubscriptionSearchRepository],
-            getattr(self, "subscription_search_repository", None),
+        queue: Optional[SubscriptionSearchRepository] = getattr(
+            self, "subscription_search_repository", None
         )
         return queue.get_batch(batch_id) if queue else None
 

app/chain/torrents.py

@@ -12,6 +12,7 @@ from app.chain.base import ChainBase
 from app.chain.media import MediaChain
 from app.domain import site as site_rules
 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.runtime.log import logger
@@ -496,6 +497,8 @@ class TorrentsChain(ChainBase):
     def _build_refresh_context(self, torrent: TorrentInfo, stype: str) -> Context:
         """识别单个种子并构造缓存上下文。"""
         logger.info(f'处理资源：{torrent.title} ...')
+        meta: MetaBase
+        mediainfo: MediaInfo | MusicInfo
         if torrent.category == MediaType.MUSIC.value:
             meta = MetaMusic.parse_query(torrent.title)
             mediainfo = MusicInfo(
@@ -508,15 +511,23 @@ class TorrentsChain(ChainBase):
             candidate_recognized = False
             match_source = "unknown"
         else:
-            meta = MetaInfo(title=torrent.title, subtitle=torrent.description)
-            if torrent.title != meta.org_string:
-                logger.info(f'种子名称应用识别词后发生改变：{torrent.title} => {meta.org_string}')
-            if meta.type != MediaType.TV and torrent.category == MediaType.TV.value:
-                meta.type = MediaType.TV
-            mediainfo = MediaChain().recognize_by_meta(meta, obtain_images=False) or MediaInfo()
-            mediainfo.clear()
-            candidate_recognized = bool(mediainfo and all(resolve_media_identity(media=mediainfo)))
-            match_source = self._get_media_id_match_source(mediainfo)
+            video_meta = MetaInfo(title=torrent.title, subtitle=torrent.description)
+            if torrent.title != video_meta.org_string:
+                logger.info(
+                    f'种子名称应用识别词后发生改变：{torrent.title} => {video_meta.org_string}'
+                )
+            if video_meta.type != MediaType.TV and torrent.category == MediaType.TV.value:
+                video_meta.type = MediaType.TV
+            video_mediainfo = (
+                MediaChain().recognize_by_meta(video_meta, obtain_images=False) or MediaInfo()
+            )
+            video_mediainfo.clear()
+            candidate_recognized = bool(
+                video_mediainfo and all(resolve_media_identity(media=video_mediainfo))
+            )
+            match_source = self._get_media_id_match_source(video_mediainfo)
+            meta = video_meta
+            mediainfo = video_mediainfo
         context = Context(
             meta_info=meta,
             media_info=mediainfo,

app/db/adapters/subscriptiondownload.py

@@ -0,0 +1,137 @@
+"""订阅下载幂等账本的短事务适配器。"""
+
+from collections.abc import Callable
+from typing import Optional, TypeVar
+
+from sqlalchemy.orm import Session
+
+from app.application.download.admission import (
+    SubscriptionDownloadClaim,
+    SubscriptionDownloadRequest,
+    SubscriptionDownloadSnapshot,
+)
+from app.db.models.subscriptiondownload import SubscriptionDownloadSubmission
+from app.db.oper.subscriptiondownload import SubscriptionDownloadOper
+from app.db.uow import SqlAlchemyUnitOfWork
+
+T = TypeVar("T")
+
+
+def _snapshot(record: SubscriptionDownloadSubmission) -> SubscriptionDownloadSnapshot:
+    """在 Session 内投影不可变提交快照。"""
+    return SubscriptionDownloadSnapshot(
+        idempotency_key=record.idempotency_key,
+        subscription_id=record.subscription_id,
+        task_id=record.task_id,
+        state=record.state,
+        attempt_count=record.attempt_count,
+        attempt_token=record.attempt_token,
+        downloader=record.downloader,
+        download_hash=record.download_hash,
+        available_at=record.available_at,
+        last_error=record.last_error,
+        created_at=record.created_at,
+        updated_at=record.updated_at,
+    )
+
+
+class TransactionalSubscriptionDownloadRepository:
+    """为每次提交认领和状态变更创建独立短事务。"""
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """保存由组合根提供的同步 Session 工厂。"""
+        self._session_factory = session_factory
+
+    def _read(self, operation: Callable[[SubscriptionDownloadOper], T]) -> T:
+        """在短 Session 中执行一次只读查询。"""
+        with self._session_factory() as session:
+            return operation(SubscriptionDownloadOper(session))
+
+    def _write(self, operation: Callable[[SubscriptionDownloadOper], T]) -> T:
+        """在显式短事务中执行一次状态变更。"""
+        with self._session_factory() as session:
+            unit_of_work = SqlAlchemyUnitOfWork(session)
+            try:
+                result = operation(SubscriptionDownloadOper(session))
+                unit_of_work.commit()
+                return result
+            except Exception:
+                unit_of_work.rollback()
+                raise
+
+    def claim(self, request: SubscriptionDownloadRequest) -> SubscriptionDownloadClaim:
+        """认领唯一提交权并返回脱离 Session 的当前状态。"""
+        def operation(repository: SubscriptionDownloadOper) -> SubscriptionDownloadClaim:
+            """在同一事务内完成唯一插入或 fenced 重领。"""
+            record, acquired = repository.claim(request)
+            return SubscriptionDownloadClaim(acquired=acquired, snapshot=_snapshot(record))
+
+        return self._write(operation)
+
+    def mark_accepted(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        downloader: Optional[str],
+        download_hash: str,
+    ) -> bool:
+        """记录下载器已接受任务。"""
+        return self._write(lambda repository: repository.mark_accepted(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            downloader=downloader,
+            download_hash=download_hash,
+        ))
+
+    def mark_succeeded(self, *, idempotency_key: str, attempt_token: str) -> bool:
+        """记录 canonical 本地结算已完成。"""
+        return self._write(lambda repository: repository.mark_succeeded(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+        ))
+
+    def mark_retryable(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        available_at: str,
+        error: Optional[str],
+    ) -> bool:
+        """记录未产生外部副作用的延迟重试状态。"""
+        return self._write(lambda repository: repository.mark_retryable(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            available_at=available_at,
+            error=error,
+        ))
+
+    def mark_reconcile_required(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        error: Optional[str],
+        downloader: Optional[str] = None,
+        download_hash: Optional[str] = None,
+    ) -> bool:
+        """冻结可能已被下载器接受的提交。"""
+        return self._write(lambda repository: repository.mark_reconcile_required(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            error=error,
+            downloader=downloader,
+            download_hash=download_hash,
+        ))
+
+    def mark_cancelled(self, *, idempotency_key: str, attempt_token: str) -> bool:
+        """在外部提交前收口取消。"""
+        return self._write(lambda repository: repository.mark_cancelled(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+        ))
+
+    def has_started_for_task(self, task_id: str) -> bool:
+        """查询搜索任务是否已越过可安全取消边界。"""
+        return self._read(lambda repository: repository.has_started_for_task(task_id))

app/db/adapters/subscriptionsearch.py

@@ -12,7 +12,6 @@ from app.application.subscription.execution import (
 )
 from app.application.subscription.sitebudget import SiteBudgetClaim
 from app.db.models.subscriptionsearch import (
-    SubscriptionSiteBudget,
     SubscriptionSearchBatch,
     SubscriptionSearchTask,
 )

app/db/models/__init__.py

@@ -5,7 +5,6 @@ from typing import Any
 
 from . import _identity  # noqa: F401  注册全局媒体身份写入不变量
 
-
 _MODEL_EXPORTS = {
     "AgentChat": ("app.db.models.agentchat", "AgentChat"),
     "AgentTask": ("app.db.models.agenttask", "AgentTask"),
@@ -47,6 +46,10 @@ _MODEL_EXPORTS = {
         "app.db.models.subscriptionsearch",
         "SubscriptionSiteBudget",
     ),
+    "SubscriptionDownloadSubmission": (
+        "app.db.models.subscriptiondownload",
+        "SubscriptionDownloadSubmission",
+    ),
     "SystemConfig": ("app.db.models.systemconfig", "SystemConfig"),
     "TransferHistory": ("app.db.models.transferhistory", "TransferHistory"),
     "TransferExecutionStep": (

app/db/models/subscriptiondownload.py

@@ -0,0 +1,52 @@
+"""订阅下载提交幂等账本模型。"""
+
+from typing import Optional
+
+from sqlalchemy import Index, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base, get_id_column
+
+
+class SubscriptionDownloadSubmission(Base):
+    """记录订阅下载唯一提交权、下载器接受事实和待对账终态。"""
+
+    id = get_id_column()
+    idempotency_key: Mapped[str] = mapped_column(String(64), nullable=False)
+    subscription_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    task_id: Mapped[Optional[str]] = mapped_column(String(64))
+    logical_identity: Mapped[str] = mapped_column(Text, nullable=False)
+    resource_key: Mapped[str] = mapped_column(Text, nullable=False)
+    coverage: Mapped[str] = mapped_column(Text, nullable=False)
+    mode: Mapped[str] = mapped_column(String(32), nullable=False)
+    state: Mapped[str] = mapped_column(String(32), nullable=False)
+    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    attempt_token: Mapped[Optional[str]] = mapped_column(String(64))
+    downloader: Mapped[Optional[str]] = mapped_column(String(128))
+    download_hash: Mapped[Optional[str]] = mapped_column(String(256))
+    available_at: Mapped[Optional[str]] = mapped_column(String(40))
+    last_error: Mapped[Optional[str]] = mapped_column(Text)
+    created_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    updated_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    started_at: Mapped[Optional[str]] = mapped_column(String(40))
+    finished_at: Mapped[Optional[str]] = mapped_column(String(40))
+
+    __table_args__ = (
+        UniqueConstraint(
+            "idempotency_key",
+            name="uq_subscriptiondownloadsubmission_idempotency_key",
+        ),
+        Index(
+            "ix_subscriptiondownloadsubmission_task_state",
+            "task_id",
+            "state",
+            "id",
+        ),
+        Index(
+            "ix_subscriptiondownloadsubmission_subscription_state",
+            "subscription_id",
+            "state",
+            "updated_at",
+            "id",
+        ),
+    )

app/db/oper/subscriptiondownload.py

@@ -0,0 +1,231 @@
+"""订阅下载提交账本的事务内状态转换。"""
+
+from datetime import datetime, timezone
+from typing import Optional
+from uuid import uuid4
+
+from sqlalchemy import and_, or_, select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.application.download.admission import SubscriptionDownloadRequest
+from app.db.base import DbOper, execute_dml
+from app.db.models.subscriptiondownload import SubscriptionDownloadSubmission
+
+
+def utc_now_text() -> str:
+    """返回可按字符串稳定排序的 UTC ISO 时间。"""
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+class SubscriptionDownloadOper(DbOper):
+    """在调用方事务内维护唯一下载提交及其 fenced 终态。"""
+
+    def claim(self, request: SubscriptionDownloadRequest) -> tuple[SubscriptionDownloadSubmission, bool]:
+        """创建或原子重领允许重试的提交记录。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅下载提交认领需要调用方提供同步 Session")
+        now = utc_now_text()
+        attempt_token = uuid4().hex
+        record = self._db.execute(
+            select(SubscriptionDownloadSubmission).where(
+                SubscriptionDownloadSubmission.idempotency_key == request.idempotency_key
+            )
+        ).scalars().first()
+        if record is None:
+            try:
+                with self._db.begin_nested():
+                    record = SubscriptionDownloadSubmission(
+                        idempotency_key=request.idempotency_key,
+                        subscription_id=request.subscription_id,
+                        task_id=request.task_id,
+                        logical_identity=request.logical_identity,
+                        resource_key=request.resource_key,
+                        coverage=request.coverage,
+                        mode=request.mode,
+                        state="submitting",
+                        attempt_count=1,
+                        attempt_token=attempt_token,
+                        created_at=now,
+                        updated_at=now,
+                        started_at=now,
+                    )
+                    self._db.add(record)
+                    self._db.flush()
+                return record, True
+            except IntegrityError:
+                self._db.expire_all()
+                record = self._db.execute(
+                    select(SubscriptionDownloadSubmission).where(
+                        SubscriptionDownloadSubmission.idempotency_key == request.idempotency_key
+                    )
+                ).scalar_one()
+
+        reclaimable = record.state in {"retryable", "cancelled"} and (
+            record.available_at is None or record.available_at <= now
+        )
+        if not reclaimable:
+            return record, False
+        updated = execute_dml(
+            self._db,
+            update(SubscriptionDownloadSubmission)
+            .where(
+                SubscriptionDownloadSubmission.id == record.id,
+                SubscriptionDownloadSubmission.state.in_(("retryable", "cancelled")),
+                or_(
+                    SubscriptionDownloadSubmission.available_at.is_(None),
+                    SubscriptionDownloadSubmission.available_at <= now,
+                ),
+            )
+            .values(
+                state="submitting",
+                task_id=request.task_id,
+                attempt_count=SubscriptionDownloadSubmission.attempt_count + 1,
+                attempt_token=attempt_token,
+                downloader=None,
+                download_hash=None,
+                available_at=None,
+                last_error=None,
+                updated_at=now,
+                started_at=now,
+                finished_at=None,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+        self._db.flush()
+        self._db.expire_all()
+        current = self._db.execute(
+            select(SubscriptionDownloadSubmission).where(
+                SubscriptionDownloadSubmission.id == record.id
+            )
+        ).scalar_one()
+        return current, bool(updated and current.attempt_token == attempt_token)
+
+    def mark_accepted(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        downloader: Optional[str],
+        download_hash: str,
+    ) -> bool:
+        """把当前 submitting 尝试推进为下载器已接受。"""
+        return self._transition(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            source_states=("submitting",),
+            values={
+                "state": "accepted",
+                "downloader": downloader,
+                "download_hash": download_hash,
+            },
+        )
+
+    def mark_succeeded(self, *, idempotency_key: str, attempt_token: str) -> bool:
+        """把当前 accepted 尝试推进为本地结算成功。"""
+        return self._transition(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            source_states=("accepted",),
+            values={"state": "succeeded", "finished_at": utc_now_text()},
+        )
+
+    def mark_retryable(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        available_at: str,
+        error: Optional[str],
+    ) -> bool:
+        """把下载器明确拒绝的当前尝试收口为延迟可重试。"""
+        return self._transition(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            source_states=("submitting",),
+            values={
+                "state": "retryable",
+                "available_at": available_at,
+                "last_error": error,
+                "finished_at": utc_now_text(),
+            },
+        )
+
+    def mark_reconcile_required(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        error: Optional[str],
+        downloader: Optional[str],
+        download_hash: Optional[str],
+    ) -> bool:
+        """冻结 submitting/accepted 尝试，禁止自动重新提交。"""
+        values: dict[str, object] = {
+            "state": "reconcile_required",
+            "last_error": error,
+            "finished_at": utc_now_text(),
+        }
+        if downloader is not None:
+            values["downloader"] = downloader
+        if download_hash is not None:
+            values["download_hash"] = download_hash
+        return self._transition(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            source_states=("submitting", "accepted"),
+            values=values,
+        )
+
+    def mark_cancelled(self, *, idempotency_key: str, attempt_token: str) -> bool:
+        """在外部提交前把当前尝试收口为已取消。"""
+        return self._transition(
+            idempotency_key=idempotency_key,
+            attempt_token=attempt_token,
+            source_states=("submitting",),
+            values={"state": "cancelled", "finished_at": utc_now_text()},
+        )
+
+    def has_started_for_task(self, task_id: str) -> bool:
+        """判断任务是否进入过外部副作用边界或已成功。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅下载提交查询需要调用方提供同步 Session")
+        return bool(self._db.execute(
+            select(SubscriptionDownloadSubmission.id)
+            .where(
+                SubscriptionDownloadSubmission.task_id == task_id,
+                SubscriptionDownloadSubmission.state.in_((
+                    "submitting",
+                    "accepted",
+                    "succeeded",
+                    "reconcile_required",
+                )),
+            )
+            .limit(1)
+        ).scalar())
+
+    def _transition(
+        self,
+        *,
+        idempotency_key: str,
+        attempt_token: str,
+        source_states: tuple[str, ...],
+        values: dict[str, object],
+    ) -> bool:
+        """以当前尝试令牌原子推进状态，拒绝过期执行者写回。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("订阅下载提交写入需要调用方提供同步 Session")
+        now = utc_now_text()
+        return bool(execute_dml(
+            self._db,
+            update(SubscriptionDownloadSubmission)
+            .where(
+                and_(
+                    SubscriptionDownloadSubmission.idempotency_key == idempotency_key,
+                    SubscriptionDownloadSubmission.attempt_token == attempt_token,
+                    SubscriptionDownloadSubmission.state.in_(source_states),
+                )
+            )
+            .values(updated_at=now, **values),
+            execution_options={"synchronize_session": False},
+        ))

app/db/oper/subscriptionsearch.py

@@ -10,9 +10,9 @@ from sqlalchemy.orm import Session
 
 from app.db.base import DbOper, execute_dml
 from app.db.models.subscriptionsearch import (
-    SubscriptionSiteBudget,
     SubscriptionSearchBatch,
     SubscriptionSearchTask,
+    SubscriptionSiteBudget,
 )
 
 
@@ -198,12 +198,13 @@ class SubscriptionSearchOper(DbOper):
             )
             self._db.flush()
             self._db.expire_all()
-            return self._db.execute(
+            claimed_task: Optional[SubscriptionSearchTask] = self._db.execute(
                 select(SubscriptionSearchTask).where(
                     SubscriptionSearchTask.id == candidate.id,
                     SubscriptionSearchTask.lease_token == lease_token,
                 )
             ).scalars().first()
+            return claimed_task
         return None
 
     def finish_task(
@@ -359,9 +360,10 @@ class SubscriptionSearchOper(DbOper):
         """按稳定批次 ID 读取聚合记录。"""
         if not isinstance(self._db, Session):
             raise RuntimeError("订阅搜索批次查询需要调用方提供同步 Session")
-        return self._db.execute(
+        batch: Optional[SubscriptionSearchBatch] = self._db.execute(
             select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id)
         ).scalars().first()
+        return batch
 
     def claim_site(
         self,
@@ -477,7 +479,10 @@ class SubscriptionSearchOper(DbOper):
 
     def _ensure_site_budget(self, *, site_id: int, now: str) -> SubscriptionSiteBudget:
         """并发安全地创建站点预算初始记录。"""
-        record = self._db.execute(
+        db = self._db
+        if not isinstance(db, Session):
+            raise RuntimeError("订阅站点预算初始化需要调用方提供同步 Session")
+        record: Optional[SubscriptionSiteBudget] = db.execute(
             select(SubscriptionSiteBudget).where(
                 SubscriptionSiteBudget.site_id == site_id
             )
@@ -485,24 +490,28 @@ class SubscriptionSearchOper(DbOper):
         if record is not None:
             return record
         try:
-            with self._db.begin_nested():
-                self._db.add(SubscriptionSiteBudget(
+            with db.begin_nested():
+                db.add(SubscriptionSiteBudget(
                     site_id=site_id,
                     next_allowed_at=now,
                     updated_at=now,
                 ))
-                self._db.flush()
+                db.flush()
         except IntegrityError:
-            self._db.expire_all()
-        return self._db.execute(
+            db.expire_all()
+        created: SubscriptionSiteBudget = db.execute(
             select(SubscriptionSiteBudget).where(
                 SubscriptionSiteBudget.site_id == site_id
             )
         ).scalar_one()
+        return created
 
     def _batch_cancel_requested(self, batch_id: str) -> bool:
         """在当前事务中读取批次取消标记。"""
-        return bool(self._db.execute(
+        db = self._db
+        if not isinstance(db, Session):
+            raise RuntimeError("订阅搜索批次取消查询需要调用方提供同步 Session")
+        return bool(db.execute(
             select(SubscriptionSearchBatch.cancel_requested).where(
                 SubscriptionSearchBatch.batch_id == batch_id
             )
@@ -510,7 +519,10 @@ class SubscriptionSearchOper(DbOper):
 
     def _refresh_batch(self, batch_id: str, *, now: str, error: Optional[str]) -> None:
         """依据所属任务终态重新计算批次计数和聚合状态。"""
-        rows = self._db.execute(
+        db = self._db
+        if not isinstance(db, Session):
+            raise RuntimeError("订阅搜索批次刷新需要调用方提供同步 Session")
+        rows = db.execute(
             select(SubscriptionSearchTask.state, func.count())  # pylint: disable=not-callable
             .where(SubscriptionSearchTask.batch_id == batch_id)
             .group_by(SubscriptionSearchTask.state)

app/domain/context.py

@@ -1524,7 +1524,7 @@ class MediaInfo:
         dicts["media_id"] = str(self.media_id) if self.media_id is not None else None
         return dicts
 
-    def clear(self):
+    def clear(self) -> None:
         """
         去除多余数据，减小体积
         """

app/startup/composition/chain.py

@@ -36,6 +36,7 @@ from app.chain.transfer.filter import (
 from app.db.adapters.chain import TransactionalChainDurableEventWriter
 from app.db.adapters.download import TransactionalDownloadFailureRepository
 from app.db.adapters.mediaserver import TransactionalMediaServerRepository
+from app.db.adapters.subscriptiondownload import TransactionalSubscriptionDownloadRepository
 from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository
 from app.db.oper.message import MessageOper
 from app.db.oper.systemconfig import SystemConfigOper
@@ -111,6 +112,7 @@ def build_chain_runtime_context(
         transfer_execution_repository=dependencies.transfer_execution,
         media_server_repository=TransactionalMediaServerRepository(SessionFactory),
         download_failure_repository=TransactionalDownloadFailureRepository(SessionFactory),
+        subscription_download_repository=TransactionalSubscriptionDownloadRepository(SessionFactory),
         user_repository=build_transactional_user_repository(),
         legacy_transfer_command=execute_legacy_transfer_command,
         configuration=configuration(),

app/startup/composition/runtime.py

@@ -48,8 +48,8 @@ from app.startup.composition.context import (
 )
 
 if TYPE_CHECKING:
-    from app.application.subscription.execution import SubscriptionSearchRepository
     from app.application.messaging.message import MessageHelper, MessageQueueManager
+    from app.application.subscription.execution import SubscriptionSearchRepository
     from app.startup.composition.agent import AgentComposition
     from app.startup.composition.configuration import ConfigurationComposition
     from app.startup.composition.database import DatabaseComposition

database/versions/e1b6d4f8a2c7_3_0_21.py

@@ -0,0 +1,72 @@
+"""3.0.21 增加订阅下载幂等与待对账提交账本。
+
+Revision ID: e1b6d4f8a2c7
+Revises: d2a7c5e9f1b4
+Create Date: 2026-09-01
+"""
+
+# Alembic 的 op 是运行期代理，静态分析无法看到实际操作方法。
+# pylint: disable=no-member
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "e1b6d4f8a2c7"
+down_revision = "d2a7c5e9f1b4"
+branch_labels = None
+depends_on = None
+
+_TABLE = "subscriptiondownloadsubmission"
+
+
+def _table_names() -> set[str]:
+    """返回当前数据库表名集合。"""
+    return set(sa.inspect(op.get_bind()).get_table_names())
+
+
+def upgrade() -> None:
+    """创建向后兼容的独立订阅下载提交账本。"""
+    if _TABLE in _table_names():
+        return
+    op.create_table(
+        _TABLE,
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("idempotency_key", sa.String(length=64), nullable=False),
+        sa.Column("subscription_id", sa.Integer(), nullable=False),
+        sa.Column("task_id", sa.String(length=64), nullable=True),
+        sa.Column("logical_identity", sa.Text(), nullable=False),
+        sa.Column("resource_key", sa.Text(), nullable=False),
+        sa.Column("coverage", sa.Text(), nullable=False),
+        sa.Column("mode", sa.String(length=32), nullable=False),
+        sa.Column("state", sa.String(length=32), nullable=False),
+        sa.Column("attempt_count", sa.Integer(), nullable=False),
+        sa.Column("attempt_token", sa.String(length=64), nullable=True),
+        sa.Column("downloader", sa.String(length=128), nullable=True),
+        sa.Column("download_hash", sa.String(length=256), nullable=True),
+        sa.Column("available_at", sa.String(length=40), nullable=True),
+        sa.Column("last_error", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.String(length=40), nullable=False),
+        sa.Column("updated_at", sa.String(length=40), nullable=False),
+        sa.Column("started_at", sa.String(length=40), nullable=True),
+        sa.Column("finished_at", sa.String(length=40), nullable=True),
+        sa.UniqueConstraint(
+            "idempotency_key",
+            name="uq_subscriptiondownloadsubmission_idempotency_key",
+        ),
+    )
+    op.create_index(
+        "ix_subscriptiondownloadsubmission_task_state",
+        _TABLE,
+        ["task_id", "state", "id"],
+    )
+    op.create_index(
+        "ix_subscriptiondownloadsubmission_subscription_state",
+        _TABLE,
+        ["subscription_id", "state", "updated_at", "id"],
+    )
+
+
+def downgrade() -> None:
+    """移除订阅下载提交账本，不改动旧下载历史。"""
+    if _TABLE in _table_names():
+        op.drop_table(_TABLE)

docs/architecture-overview.md

@@ -754,8 +754,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 927 |
-| 内部导入边 | 7,745 |
+| Python 模块 | 932 |
+| 内部导入边 | 7,784 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

docs/architecture/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 927 / 7,745 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 932 / 7,784 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -102,8 +102,8 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 |
 | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
-| 全量 mypy 历史债务 | 9,603 / 517 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
-| Ruff 历史诊断 | 569 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| 全量 mypy 历史债务 | 9,589 / 517 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
+| Ruff 历史诊断 | 568 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件

docs/refactor/moviepilot-subscription-governance-roadmap.md

@@ -1,7 +1,7 @@
 # MoviePilot 订阅执行治理
 
 > 状态：`active（2026-09-01 已由 MoviePilot v3 接管）`
-> 当前叶：`SUB-GOV-003A`
+> 当前叶：`SUB-GOV-003B`
 > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、
 > `V3-RDY-009A1C`、`V3-RDY-009A1D`
 > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环
@@ -235,11 +235,11 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
 | `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `completed（2026-09-01）` |
 | `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `completed（2026-09-01）` |
 | `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `completed（2026-09-01）` |
-| `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `in_progress` |
-| `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` |
+| `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `completed（2026-09-01）` |
+| `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `in_progress` |
 | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` |
 
-当前只激活 `SUB-GOV-003A`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
+当前只激活 `SUB-GOV-003B`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B
 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。
 
 ### 6.1 SUB-GOV-001A 验收证据
@@ -343,6 +343,23 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收
 - 验证：联合队列/站点预算/搜索状态 `35 passed`，订阅/搜索/调度架构 `58 passed`，错误级 Pylint 为 0，
   Host 架构基线与 `git diff --check` 通过。
 
+### 6.9 SUB-GOV-003A 验收证据
+
+- 新增持久 `SubscriptionDownloadSubmission` 账本和 `SubscriptionDownloadGovernance` 应用合同；幂等键包含
+  订阅 ID、媒体逻辑身份、torrent 稳定身份、精确季集覆盖和普通/洗版模式，唯一约束覆盖 RSS、Spider、
+  手工搜索与兜底搜索的重叠入口；
+- 下载器调用前以 fenced attempt token 原子认领；明确拒绝进入带冷却的 retryable，已接受但本地结算失败或
+  下载器超时进入 `reconcile_required`，不得自动重试；成功记录可复用既有 hash，迁移前同订阅同 torrent
+  同覆盖的下载历史也参与兼容去重；
+- 生产订阅下载路径在提交前重新读取订阅、缺集、下载历史、洗版优先级和当前下载参数；订阅删除、暂停或筛选
+  条件变化时停止提交，准备阶段快照不越过外部副作用边界；
+- 取消在下载器调用前写入 cancelled 且不产生外部提交；越过提交边界后的取消按真实下载结果完成并保留说明，
+  不伪造 cancelled 终态；任务重启通过持久账本判断是否已经开始；
+- Alembic `e1b6d4f8a2c7` 为唯一 head，迁移支持 SQLite/PostgreSQL、可逆且重复执行安全；公开
+  `DownloadChain.download_single()` / `batch_download()` 只在末尾增加可选治理参数，旧插件 ABI 保持兼容；
+- 验证：下载幂等、取消补偿、订阅提交与搜索治理 `131 passed`；架构合同、依赖、复杂度和质量 ratchet
+  `284 passed`；Ruff 低水位 `568`、mypy 低水位 `9,589`，Host 架构基线与 `git diff --check` 通过。
+
 ## 7. 上线前验证与验收
 
 ### 7.1 场景

tests/conftest.py

@@ -283,6 +283,7 @@ def configure_plugin_system_services():
     from app.db.adapters.mediaserver import TransactionalMediaServerRepository
     from app.db.adapters.site import TransactionalSiteRepository
     from app.db.adapters.subscription import TransactionalSubscriptionRepository
+    from app.db.adapters.subscriptiondownload import TransactionalSubscriptionDownloadRepository
     from app.db.adapters.transaction import TransactionalWriteRunner
     from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository
     from app.db.adapters.transfer.execution import (
@@ -410,6 +411,7 @@ def configure_plugin_system_services():
             transfer_execution_repository=TransactionalTransferExecutionRepository(SessionFactory),
             media_server_repository=TransactionalMediaServerRepository(SessionFactory),
             download_failure_repository=TransactionalDownloadFailureRepository(SessionFactory),
+            subscription_download_repository=TransactionalSubscriptionDownloadRepository(SessionFactory),
             user_repository=user_repository(),
             configuration=build_chain_runtime_config(settings),
         )

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 7745,
-  "edge_sha256": "4c722884c3a228e2fc51d5389d4fec3e10436e797c4ae168b9306154a9179475",
+  "edge_count": 7784,
+  "edge_sha256": "602f1b401b10d5a26a0782fee6ca31b1276dd9861ec9006be35d1372c4974e59",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -3694,8 +3694,21 @@
     "app.chain.douban -> app.schemas",
     "app.chain.douban -> app.schemas.context",
     "app.chain.douban -> app.schemas.types",
+    "app.chain.download.admission -> app.application",
+    "app.chain.download.admission -> app.application.download",
+    "app.chain.download.admission -> app.application.download.admission",
+    "app.chain.download.admission -> app.chain",
+    "app.chain.download.admission -> app.chain.download",
+    "app.chain.download.admission -> app.chain.download.contract",
+    "app.chain.download.admission -> app.domain",
+    "app.chain.download.admission -> app.domain.context",
+    "app.chain.download.admission -> app.domain.episode",
+    "app.chain.download.admission -> app.schemas",
+    "app.chain.download.admission -> app.schemas.media",
+    "app.chain.download.admission -> app.schemas.types",
     "app.chain.download.batch -> app.application",
     "app.chain.download.batch -> app.application.download",
+    "app.chain.download.batch -> app.application.download.admission",
     "app.chain.download.batch -> app.application.download.selection",
     "app.chain.download.batch -> app.application.torrent",
     "app.chain.download.batch -> app.application.torrent.download",
@@ -3729,6 +3742,7 @@
     "app.chain.download.existence -> app.schemas.types",
     "app.chain.download.facade -> app.chain",
     "app.chain.download.facade -> app.chain.download",
+    "app.chain.download.facade -> app.chain.download.admission",
     "app.chain.download.facade -> app.chain.download.batch",
     "app.chain.download.facade -> app.chain.download.existence",
     "app.chain.download.facade -> app.chain.download.failure",
@@ -3793,6 +3807,7 @@
     "app.chain.download.selection -> app.application",
     "app.chain.download.selection -> app.application.configuration",
     "app.chain.download.selection -> app.application.download",
+    "app.chain.download.selection -> app.application.download.admission",
     "app.chain.download.selection -> app.application.download.failures",
     "app.chain.download.selection -> app.application.torrent",
     "app.chain.download.selection -> app.application.torrent.download",
@@ -3815,6 +3830,8 @@
     "app.chain.download.submission -> app.application",
     "app.chain.download.submission -> app.application.configuration",
     "app.chain.download.submission -> app.application.directory",
+    "app.chain.download.submission -> app.application.download",
+    "app.chain.download.submission -> app.application.download.admission",
     "app.chain.download.submission -> app.application.torrent",
     "app.chain.download.submission -> app.application.torrent.download",
     "app.chain.download.submission -> app.chain",
@@ -4521,6 +4538,8 @@
     "app.chain.subscribe.notify -> app.schemas.message",
     "app.chain.subscribe.notify -> app.schemas.types",
     "app.chain.subscribe.policy -> app.application",
+    "app.chain.subscribe.policy -> app.application.download",
+    "app.chain.subscribe.policy -> app.application.download.admission",
     "app.chain.subscribe.policy -> app.application.subscription",
     "app.chain.subscribe.policy -> app.application.subscription.contract",
     "app.chain.subscribe.policy -> app.application.subscription.priority",
@@ -4656,6 +4675,7 @@
     "app.chain.torrents -> app.domain",
     "app.chain.torrents -> app.domain.context",
     "app.chain.torrents -> app.domain.meta",
+    "app.chain.torrents -> app.domain.meta.metabase",
     "app.chain.torrents -> app.domain.meta.metamusic",
     "app.chain.torrents -> app.domain.metainfo",
     "app.chain.torrents -> app.domain.site",
@@ -5146,6 +5166,15 @@
     "app.db.adapters.subscription -> app.schemas",
     "app.db.adapters.subscription -> app.schemas.common",
     "app.db.adapters.subscription -> app.schemas.types",
+    "app.db.adapters.subscriptiondownload -> app.application",
+    "app.db.adapters.subscriptiondownload -> app.application.download",
+    "app.db.adapters.subscriptiondownload -> app.application.download.admission",
+    "app.db.adapters.subscriptiondownload -> app.db",
+    "app.db.adapters.subscriptiondownload -> app.db.models",
+    "app.db.adapters.subscriptiondownload -> app.db.models.subscriptiondownload",
+    "app.db.adapters.subscriptiondownload -> app.db.oper",
+    "app.db.adapters.subscriptiondownload -> app.db.oper.subscriptiondownload",
+    "app.db.adapters.subscriptiondownload -> app.db.uow",
     "app.db.adapters.subscriptionsearch -> app.application",
     "app.db.adapters.subscriptionsearch -> app.application.subscription",
     "app.db.adapters.subscriptionsearch -> app.application.subscription.execution",
@@ -5294,6 +5323,8 @@
     "app.db.models.subscribehistory -> app.db.models._constraints",
     "app.db.models.subscribehistory -> app.schemas",
     "app.db.models.subscribehistory -> app.schemas.types",
+    "app.db.models.subscriptiondownload -> app.db",
+    "app.db.models.subscriptiondownload -> app.db.base",
     "app.db.models.subscriptionsearch -> app.db",
     "app.db.models.subscriptionsearch -> app.db.base",
     "app.db.models.systemconfig -> app.db",
@@ -5402,6 +5433,13 @@
     "app.db.oper.subscribehistory -> app.schemas.common",
     "app.db.oper.subscribehistory -> app.schemas.query",
     "app.db.oper.subscribehistory -> app.schemas.types",
+    "app.db.oper.subscriptiondownload -> app.application",
+    "app.db.oper.subscriptiondownload -> app.application.download",
+    "app.db.oper.subscriptiondownload -> app.application.download.admission",
+    "app.db.oper.subscriptiondownload -> app.db",
+    "app.db.oper.subscriptiondownload -> app.db.base",
+    "app.db.oper.subscriptiondownload -> app.db.models",
+    "app.db.oper.subscriptiondownload -> app.db.models.subscriptiondownload",
     "app.db.oper.subscriptionsearch -> app.db",
     "app.db.oper.subscriptionsearch -> app.db.base",
     "app.db.oper.subscriptionsearch -> app.db.models",
@@ -8044,6 +8082,7 @@
     "app.startup.composition.chain -> app.db.adapters.chain",
     "app.startup.composition.chain -> app.db.adapters.download",
     "app.startup.composition.chain -> app.db.adapters.mediaserver",
+    "app.startup.composition.chain -> app.db.adapters.subscriptiondownload",
     "app.startup.composition.chain -> app.db.adapters.transfer",
     "app.startup.composition.chain -> app.db.adapters.transfer.admission",
     "app.startup.composition.chain -> app.db.oper",
@@ -8838,7 +8877,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 927,
+  "module_count": 932,
   "modules": [
     "app",
     "app.adapters",
@@ -9042,6 +9081,7 @@
     "app.application.database",
     "app.application.directory",
     "app.application.download",
+    "app.application.download.admission",
     "app.application.download.failures",
     "app.application.download.selection",
     "app.application.download.tasks",
@@ -9161,6 +9201,7 @@
     "app.chain.dashboard",
     "app.chain.douban",
     "app.chain.download",
+    "app.chain.download.admission",
     "app.chain.download.batch",
     "app.chain.download.contract",
     "app.chain.download.existence",
@@ -9266,6 +9307,7 @@
     "app.db.adapters.query",
     "app.db.adapters.site",
     "app.db.adapters.subscription",
+    "app.db.adapters.subscriptiondownload",
     "app.db.adapters.subscriptionsearch",
     "app.db.adapters.transaction",
     "app.db.adapters.transfer",
@@ -9300,6 +9342,7 @@
     "app.db.models.siteuserdata",
     "app.db.models.subscribe",
     "app.db.models.subscribehistory",
+    "app.db.models.subscriptiondownload",
     "app.db.models.subscriptionsearch",
     "app.db.models.systemconfig",
     "app.db.models.transferexecutionstep",
@@ -9323,6 +9366,7 @@
     "app.db.oper.site",
     "app.db.oper.subscribe",
     "app.db.oper.subscribehistory",
+    "app.db.oper.subscriptiondownload",
     "app.db.oper.subscriptionsearch",
     "app.db.oper.systemconfig",
     "app.db.oper.transferexecutionstep",

tests/fixtures/architecture/mypy-baseline.json

@@ -968,8 +968,8 @@
   },
   "app/chain/subscribe/match.py": {
     "arg-type": 4,
-    "assignment": 2,
-    "no-untyped-call": 4,
+    "assignment": 1,
+    "no-untyped-call": 1,
     "union-attr": 7
   },
   "app/chain/subscribe/notify.py": {
@@ -990,7 +990,7 @@
   "app/chain/subscribe/refresh.py": {
     "arg-type": 3,
     "assignment": 2,
-    "no-redef": 2,
+    "no-redef": 1,
     "var-annotated": 2
   },
   "app/chain/subscribe/search.py": {
@@ -1009,9 +1009,8 @@
     "type-arg": 2
   },
   "app/chain/torrents.py": {
-    "arg-type": 28,
-    "assignment": 4,
-    "no-any-return": 1,
+    "arg-type": 22,
+    "assignment": 3,
     "no-untyped-call": 1,
     "no-untyped-def": 15,
     "type-arg": 5,
@@ -1275,7 +1274,7 @@
     "assignment": 86,
     "no-any-return": 1,
     "no-untyped-call": 4,
-    "no-untyped-def": 24,
+    "no-untyped-def": 23,
     "operator": 1,
     "type-arg": 36
   },

tests/fixtures/architecture/ruff-baseline.json

@@ -96,9 +96,6 @@
   "app/db/diagnostics.py": {
     "I001": 1
   },
-  "app/db/models/__init__.py": {
-    "I001": 1
-  },
   "app/db/models/agentchat.py": {
     "I001": 1
   },

tests/test_download_architecture.py

@@ -39,6 +39,7 @@ def test_download_monolith_is_retired_and_package_owners_are_complete() -> None:
     assert not (CHAIN_ROOT / "download.py").exists()
     assert {path.name for path in DOWNLOAD_PACKAGE.glob("*.py")} == {
         "__init__.py",
+        "admission.py",
         "batch.py",
         "contract.py",
         "existence.py",

tests/test_download_compat.py

@@ -56,6 +56,7 @@ def test_download_single_accepts_existing_plugin_call_shapes() -> None:
     )
     assert signature.parameters["return_detail"].default is False
     assert signature.parameters["custom_words"].default is None
+    assert signature.parameters["governance"].default is None
 
 
 def test_batch_download_accepts_existing_plugin_call_shapes() -> None:
@@ -73,6 +74,7 @@ def test_batch_download_accepts_existing_plugin_call_shapes() -> None:
         downloader="qbittorrent",
         custom_words="S04E05 => S01E170",
     )
+    assert signature.parameters["governance"].default is None
 
 
 def test_get_no_exists_info_accepts_existing_plugin_call_shapes() -> None:

tests/test_subscription_download_governance.py

@@ -0,0 +1,464 @@
+"""订阅下载跨入口幂等、不确定终态与取消补偿测试。"""
+
+from __future__ import annotations
+
+import importlib
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+import app.chain.download.submission as download_submission
+from app.application.download.admission import (
+    DownloadReconciliationRequired,
+    SubscriptionDownloadGovernance,
+    SubscriptionDownloadRequest,
+)
+from app.application.subscription.contract import SubscriptionSnapshot
+from app.chain.download import DownloadChain
+from app.chain.subscribe import policy as subscribe_policy
+from app.chain.subscribe.facade import SubscribeChain
+from app.db.adapters.subscriptiondownload import TransactionalSubscriptionDownloadRepository
+from app.db.base import Base
+from app.db.models.subscriptiondownload import SubscriptionDownloadSubmission
+from app.domain.context import Context, MediaInfo, TorrentInfo
+from app.domain.metainfo import MetaInfo
+from app.schemas.types import MediaSource, MediaType
+
+
+def _request(key: str = "key-1", task_id: str | None = "task-1") -> SubscriptionDownloadRequest:
+    """构造固定身份的提交认领请求。"""
+    return SubscriptionDownloadRequest(
+        idempotency_key=key,
+        subscription_id=7,
+        task_id=task_id,
+        logical_identity='{"subscription_id":7}',
+        resource_key="example.com:id=42",
+        coverage="episodes:E01-E03",
+        mode="normal",
+    )
+
+
+def _repository(tmp_path) -> tuple[TransactionalSubscriptionDownloadRepository, object]:
+    """创建独立 SQLite 幂等账本仓储与 Session 工厂。"""
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'subscription-download.db'}",
+        connect_args={"check_same_thread": False},
+    )
+    Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine)
+    return TransactionalSubscriptionDownloadRepository(factory), factory
+
+
+def test_repository_claims_same_submission_once_across_workers(tmp_path) -> None:
+    """并发入口对同一幂等键只能有一个取得下载器提交权。"""
+    repository, _factory = _repository(tmp_path)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        claims = list(executor.map(lambda _index: repository.claim(_request()), range(2)))
+
+    assert sum(claim.acquired for claim in claims) == 1
+    assert {claim.snapshot.state for claim in claims} == {"submitting"}
+    assert {claim.snapshot.attempt_count for claim in claims} == {1}
+
+
+def test_repository_fences_retry_and_preserves_uncertain_terminal(tmp_path) -> None:
+    """明确拒绝可延迟重试，过期令牌和待对账状态均不得重新提交。"""
+    repository, _factory = _repository(tmp_path)
+    first = repository.claim(_request())
+    token = first.snapshot.attempt_token
+    assert token
+    assert repository.mark_retryable(
+        idempotency_key="key-1",
+        attempt_token=token,
+        available_at="1970-01-01T00:00:00+00:00",
+        error="downloader rejected",
+    )
+
+    second = repository.claim(_request(task_id="task-2"))
+    second_token = second.snapshot.attempt_token
+    assert second.acquired
+    assert second_token and second_token != token
+    assert not repository.mark_succeeded(
+        idempotency_key="key-1",
+        attempt_token=token,
+    )
+    assert repository.mark_reconcile_required(
+        idempotency_key="key-1",
+        attempt_token=second_token,
+        error="transport timeout",
+    )
+
+    blocked = repository.claim(_request(task_id="task-3"))
+    assert not blocked.acquired
+    assert blocked.snapshot.state == "reconcile_required"
+    assert blocked.snapshot.attempt_count == 2
+    assert repository.has_started_for_task("task-2")
+
+
+class _FakeTorrentHelper:
+    """返回固定种子目录和文件清单，隔离真实 bencode 解析。"""
+
+    @staticmethod
+    def get_fileinfo_from_torrent_content(_content):
+        """返回单文件种子结构。"""
+        return "Demo.Show.S01", ["Demo.Show.S01E01.mkv"]
+
+
+def _download_chain(repository) -> DownloadChain:
+    """构造只执行下载提交边界的 Chain 测试实例。"""
+    chain = DownloadChain.__new__(DownloadChain)
+    chain.subscription_download_repository = repository
+    chain.download_history_repository = MagicMock()
+    chain.download_history_repository.get_by_media_identity.return_value = []
+    chain.download_failure_repository = MagicMock()
+    chain.eventmanager = MagicMock()
+    chain.eventmanager.send_event.return_value = None
+    chain.post_message = MagicMock()
+    chain._settle_download_success = MagicMock()
+    chain._resolve_media_download_dir = MagicMock(
+        return_value=("local", Path("/downloads"), None)
+    )
+    chain.runtime_config = SimpleNamespace(media_extensions=(".mkv",))
+    return chain
+
+
+def _context() -> Context:
+    """构造具有稳定媒体和 torrent 身份的电视剧候选。"""
+    return Context(
+        meta_info=MetaInfo("Demo Show S01E01"),
+        media_info=MediaInfo(
+            media_source=MediaSource.TMDB,
+            media_id="77",
+            type=MediaType.TV,
+            title="Demo Show",
+            year="2026",
+            tmdb_id=77,
+            genre_ids=[18],
+        ),
+        torrent_info=TorrentInfo(
+            site=12,
+            site_name="TestSite",
+            title="Demo Show S01E01 1080p",
+            enclosure="https://example.com/download.php?id=42",
+        ),
+    )
+
+
+@pytest.fixture(autouse=True)
+def _submission_dependencies(monkeypatch):
+    """固定目录、媒体补全和种子解析，避免触发外部模块。"""
+    monkeypatch.setattr(
+        "app.application.directory.DirectoryHelper.get_download_dirs",
+        lambda _self: [SimpleNamespace(
+            storage="local",
+            download_path="/downloads",
+            category=None,
+            media_type=None,
+        )],
+    )
+    monkeypatch.setattr(download_submission, "TorrentHelper", _FakeTorrentHelper)
+    monkeypatch.setattr(
+        download_submission.MediaChain,
+        "supplement_tmdb_info",
+        lambda _self, media, _meta: media,
+    )
+
+
+def test_download_chain_reuses_success_without_second_downloader_call(tmp_path) -> None:
+    """重叠入口在首个提交成功后复用 hash，不再次调用下载器。"""
+    repository, _factory = _repository(tmp_path)
+    chain = _download_chain(repository)
+    chain.download = MagicMock(return_value=("qb", "hash-1", "Original", "accepted"))
+    governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal")
+
+    first = chain.download_single(
+        context=_context(),
+        torrent_content=b"torrent",
+        episodes={1},
+        save_path="/downloads",
+        governance=governance,
+    )
+    second = chain.download_single(
+        context=_context(),
+        torrent_content=b"torrent",
+        episodes={1},
+        save_path="/downloads",
+        governance=governance,
+    )
+
+    assert first == second == "hash-1"
+    chain.download.assert_called_once()
+    chain._settle_download_success.assert_called_once()
+
+
+def test_download_chain_freezes_when_local_settlement_fails(tmp_path) -> None:
+    """下载器接受而历史结算失败时进入待对账，后续入口不得盲重试。"""
+    repository, _factory = _repository(tmp_path)
+    chain = _download_chain(repository)
+    chain.download = MagicMock(return_value=("qb", "hash-2", "Original", "accepted"))
+    chain._settle_download_success.side_effect = RuntimeError("history unavailable")
+    governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal")
+
+    with pytest.raises(DownloadReconciliationRequired):
+        chain.download_single(
+            context=_context(),
+            torrent_content=b"torrent",
+            episodes={1},
+            save_path="/downloads",
+            governance=governance,
+        )
+    with pytest.raises(DownloadReconciliationRequired):
+        chain.download_single(
+            context=_context(),
+            torrent_content=b"torrent",
+            episodes={1},
+            save_path="/downloads",
+            governance=governance,
+        )
+
+    chain.download.assert_called_once()
+
+
+def test_download_chain_freezes_when_downloader_result_is_uncertain(tmp_path) -> None:
+    """下载器调用抛错可能已产生副作用，重启后的新实例仍不得自动重试。"""
+    repository, _factory = _repository(tmp_path)
+    chain = _download_chain(repository)
+    chain.download = MagicMock(side_effect=TimeoutError("response timeout"))
+    governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal")
+
+    with pytest.raises(DownloadReconciliationRequired):
+        chain.download_single(
+            context=_context(),
+            torrent_content=b"torrent",
+            episodes={1},
+            save_path="/downloads",
+            governance=governance,
+        )
+
+    restarted = _download_chain(repository)
+    restarted.download = MagicMock()
+    with pytest.raises(DownloadReconciliationRequired):
+        restarted.download_single(
+            context=_context(),
+            torrent_content=b"torrent",
+            episodes={1},
+            save_path="/downloads",
+            governance=governance,
+        )
+    restarted.download.assert_not_called()
+
+
+def test_download_chain_delays_retry_after_explicit_rejection(tmp_path) -> None:
+    """下载器明确拒绝且无 hash 时进入冷却，不把不确定和已拒绝混为一谈。"""
+    repository, factory = _repository(tmp_path)
+    chain = _download_chain(repository)
+    chain.download = MagicMock(return_value=("qb", None, "Original", "downloader rejected"))
+    governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal")
+
+    first = chain.download_single(
+        context=_context(),
+        torrent_content=b"torrent",
+        episodes={1},
+        save_path="/downloads",
+        governance=governance,
+    )
+    second = chain.download_single(
+        context=_context(),
+        torrent_content=b"torrent",
+        episodes={1},
+        save_path="/downloads",
+        governance=governance,
+    )
+
+    assert first is None and second is None
+    chain.download.assert_called_once()
+    with factory() as session:
+        record = session.query(SubscriptionDownloadSubmission).one()
+        assert record.state == "retryable"
+        assert record.available_at > record.updated_at
+
+
+def test_download_chain_reads_legacy_history_before_new_ledger(tmp_path) -> None:
+    """迁移前同订阅同 torrent 同覆盖的成功历史仍能阻止重复提交。"""
+    repository, factory = _repository(tmp_path)
+    chain = _download_chain(repository)
+    chain.download = MagicMock()
+    chain.download_history_repository.get_by_media_identity.return_value = [
+        SimpleNamespace(
+            download_hash="legacy-hash",
+            torrent_name="Demo Show S01E01 1080p",
+            torrent_site="TestSite",
+            episodes="E01",
+            seasons="S01",
+            episode_group=None,
+            note={"source": 'Subscribe|{"id": 7}'},
+        )
+    ]
+
+    result = chain.download_single(
+        context=_context(),
+        torrent_content=b"torrent",
+        episodes={1},
+        save_path="/downloads",
+        governance=SubscriptionDownloadGovernance(subscription_id=7, mode="normal"),
+    )
+
+    assert result == "legacy-hash"
+    chain.download.assert_not_called()
+    with factory() as session:
+        assert session.query(SubscriptionDownloadSubmission).count() == 0
+
+
+def test_download_chain_cancels_before_external_side_effect(tmp_path) -> None:
+    """取消在下载器边界前生效时不得创建下载任务或成功事实。"""
+    repository, _factory = _repository(tmp_path)
+    chain = _download_chain(repository)
+    chain.download = MagicMock()
+    governance = SubscriptionDownloadGovernance(
+        subscription_id=7,
+        mode="normal",
+        task_id="cancel-task",
+        cancelled=lambda: True,
+    )
+
+    result = chain.download_single(
+        context=_context(),
+        torrent_content=b"torrent",
+        episodes={1},
+        save_path="/downloads",
+        governance=governance,
+    )
+
+    assert result is None
+    chain.download.assert_not_called()
+    assert not repository.has_started_for_task("cancel-task")
+
+
+def test_subscription_policy_reloads_facts_and_threads_governance(monkeypatch) -> None:
+    """提交策略必须使用当前订阅和重算缺集，而不是直接消费准备阶段快照。"""
+    prepared = SubscriptionSnapshot(
+        id=7,
+        name="Demo Show",
+        type=MediaType.TV.value,
+        media_source=MediaSource.TMDB,
+        media_id="77",
+        season=1,
+        state="R",
+        save_path="/old",
+        downloader="old",
+    )
+    current = SubscriptionSnapshot(
+        **{
+            **prepared.to_dict(),
+            "save_path": "/current",
+            "downloader": "current",
+            "lack_episode": 1,
+        }
+    )
+    context = _context()
+    stale_missing = {"themoviedb:77": {1: SimpleNamespace(episodes=[1, 2])}}
+    fresh_missing = {"themoviedb:77": {1: SimpleNamespace(episodes=[1])}}
+    captured = {}
+
+    class _FakeDownloadChain:
+        """捕获提交策略传入的当前事实与治理上下文。"""
+
+        def batch_download(self, **kwargs):
+            """记录一次批量提交并返回当前缺集。"""
+            captured.update(kwargs)
+            return [], kwargs["no_exists"]
+
+    chain = SubscribeChain.__new__(SubscribeChain)
+    chain.subscription_repository = SimpleNamespace(get=lambda _subscribe_id: current)
+    chain.check_and_handle_existing_media = MagicMock(return_value=(False, fresh_missing))
+    monkeypatch.setattr(
+        chain,
+        "_SubscribeChain__revalidate_download_contexts",
+        lambda _subscribe, contexts: contexts,
+    )
+    monkeypatch.setattr(subscribe_policy, "DownloadChain", _FakeDownloadChain)
+
+    _downloads, lefts = chain._SubscribeChain__download_best_version_with_full_pack_first(
+        contexts=[context],
+        no_exists=stale_missing,
+        subscribe=prepared,
+        mediakey="themoviedb:77",
+        save_path="/old",
+        downloader="old",
+    )
+
+    assert lefts is fresh_missing
+    chain.check_and_handle_existing_media.assert_called_once_with(
+        subscribe=current,
+        meta=context.meta_info,
+        mediainfo=context.media_info,
+        mediakey="themoviedb:77",
+    )
+    assert captured["no_exists"] is fresh_missing
+    assert captured["save_path"] == "/current"
+    assert captured["downloader"] == "current"
+    assert captured["governance"].subscription_id == 7
+    assert captured["governance"].mode == "normal"
+
+
+def test_subscription_policy_discards_candidates_after_filter_change(monkeypatch) -> None:
+    """准备后筛选合同变化时必须放弃旧候选，不允许用旧快照提交。"""
+    prepared = SubscriptionSnapshot(
+        id=7,
+        name="Demo Show",
+        type=MediaType.TV.value,
+        media_source=MediaSource.TMDB,
+        media_id="77",
+        season=1,
+        state="R",
+        quality="1080p",
+    )
+    current = SubscriptionSnapshot(**{**prepared.to_dict(), "quality": "2160p"})
+    chain = SubscribeChain.__new__(SubscribeChain)
+    chain.subscription_repository = SimpleNamespace(get=lambda _subscribe_id: current)
+    batch_download = MagicMock(side_effect=AssertionError("stale candidate submitted"))
+    monkeypatch.setattr(subscribe_policy, "DownloadChain", lambda: SimpleNamespace(
+        batch_download=batch_download,
+    ))
+
+    downloads, lefts = chain._SubscribeChain__download_best_version_with_full_pack_first(
+        contexts=[_context()],
+        no_exists={"themoviedb:77": {}},
+        subscribe=prepared,
+        mediakey="themoviedb:77",
+    )
+
+    assert downloads == []
+    assert lefts == {"themoviedb:77": {}}
+    batch_download.assert_not_called()
+
+
+def test_subscription_download_migration_is_idempotent_and_reversible(tmp_path, monkeypatch) -> None:
+    """3.0.21 迁移可重复升级，并只移除自身新增的兼容表。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'migration.db'}")
+    migration = importlib.import_module("database.versions.e1b6d4f8a2c7_3_0_21")
+    with engine.begin() as connection:
+        operations = Operations(MigrationContext.configure(connection))
+        monkeypatch.setattr(migration, "op", operations)
+        migration.upgrade()
+        migration.upgrade()
+        inspector = sa.inspect(connection)
+        assert "subscriptiondownloadsubmission" in inspector.get_table_names()
+        indexes = {item["name"] for item in inspector.get_indexes("subscriptiondownloadsubmission")}
+        assert "ix_subscriptiondownloadsubmission_task_state" in indexes
+        migration.downgrade()
+        assert "subscriptiondownloadsubmission" not in sa.inspect(connection).get_table_names()
+
+
+def test_model_metadata_registers_submission_table() -> None:
+    """显式模型注册必须让 fresh create_all 包含订阅提交账本。"""
+    assert SubscriptionDownloadSubmission.__tablename__ == "subscriptiondownloadsubmission"
+    assert "subscriptiondownloadsubmission" in Base.metadata.tables

tests/test_subscription_governance_replay.py

@@ -14,7 +14,6 @@ from app.domain.metainfo import MetaInfo
 from app.schemas.mediaserver import NotExistMediaInfo
 from app.schemas.types import MediaSource, MediaType
 
-
 _FIXTURE_PATH = Path(__file__).parent / "fixtures" / "subscription_governance_match_replay.json"
 
 

tests/test_subscription_search_governance.py

@@ -2,6 +2,7 @@
 
 from dataclasses import replace
 from datetime import datetime, timedelta
+from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
 from sqlalchemy import create_engine
@@ -116,3 +117,30 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke
     assert batch.finished_count == 1
     assert batch.failed_count == 1
     assert batch.last_error == "provider timeout"
+
+
+def test_late_cancel_completes_when_download_submission_already_started(tmp_path, monkeypatch):
+    """取消晚于下载器副作用边界时按真实结果完成，不能伪装成未执行取消。"""
+    subscribe = _subscribe(5)
+    chain = _chain(tmp_path, [subscribe])
+    monkeypatch.setattr(chain, "_search_batch_available_at", lambda _source: "1970-01-01T00:00:00+00:00")
+    chain.subscription_download_repository = SimpleNamespace(
+        has_started_for_task=lambda _task_id: False,
+    )
+    queue = chain.subscription_search_repository
+    cancel_checks = iter((False, True))
+    monkeypatch.setattr(queue, "is_cancel_requested", lambda _task_id: next(cancel_checks))
+    def process(item, _searchchain):
+        """模拟当前任务复用或完成下载后才收到取消。"""
+        chain._mark_subscription_download_started()
+        return item
+
+    monkeypatch.setattr(chain, "_process_search_subscription", process)
+
+    with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()):
+        batch_id = chain.search(state="R")
+
+    batch = chain.get_search_batch(batch_id)
+    assert batch.state == "completed"
+    assert batch.finished_count == 1
+    assert batch.cancelled_count == 0

tests/test_subscription_site_budget.py

@@ -15,10 +15,10 @@ from app.application.subscription.sitebudget import (
     SubscriptionSearchCancelled,
     SubscriptionSiteBudget,
 )
+from app.chain.search.facade import SearchChain
 from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository
 from app.db.base import Base
 from app.db.models.subscriptionsearch import SubscriptionSiteBudget as SiteBudgetRecord
-from app.chain.search.facade import SearchChain
 from app.modules.indexer import _classify_search_failure
 from app.runtime.stop import ProcessStopState