refactor: add durable transfer planning checkpoints

app/application/chain/context.py

@@ -11,9 +11,9 @@ from app.application.chain.durable_events import ChainDurableEventWriter
 from app.application.configuration import ChainRuntimeConfig
 from app.runtime.stop import StopState, runtime_stop_state
 
-
 MessageQueueFactory = Callable[[Callable[..., Any]], Any]
 ModuleDispatcherFactory = Callable[..., Any]
+LegacyTransferCommand = Callable[..., Any]
 ChainRuntimeContextProvider = Callable[[], "ChainRuntimeContext"]
 
 
@@ -30,6 +30,7 @@ class ChainRuntimeContext:
     async_file_cache: Any
     message_queue_factory: MessageQueueFactory
     module_dispatcher_factory: ModuleDispatcherFactory
+    legacy_transfer_command: Optional[LegacyTransferCommand] = None
     data_ports: Optional[ChainDataPorts] = None
     durable_event_writer: Optional[ChainDurableEventWriter] = None
     configuration: ChainRuntimeConfig = field(

app/application/transfer.py

@@ -14,12 +14,24 @@ TransferJob / TransferJobTask，那两个用 app.schemas 的同名 DTO——一
 视图，分开表达之后两边都不必再迁就对方。
 """
 import asyncio
+import hashlib
+import json
 import threading
 from copy import deepcopy
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from time import monotonic
-from typing import Callable, Dict, List, Optional, Protocol, Tuple, Union
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Protocol,
+    Tuple,
+    TypeAlias,
+    Union,
+)
 
 from pydantic import BaseModel, ConfigDict, PrivateAttr
 
@@ -49,6 +61,571 @@ from app.schemas.types import (
 )
 from app.schemas.workflow import MediaInfo as _SchemaMediaInfo
 
+JSONValue: TypeAlias = Union[None, bool, int, float, str, list["JSONValue"], dict[str, "JSONValue"]]
+
+TRANSFER_ADMISSION_ACCEPTED = "accepted"
+TRANSFER_ADMISSION_PROVIDER_PENDING = "provider_pending"
+TRANSFER_ADMISSION_PLANNED = "planned"
+TRANSFER_PLANNING_INPUT_VERSION = 1
+TRANSFER_PLAN_CHECKPOINT_VERSION = 1
+TRANSFER_PROVIDER_INVOCATION_VERSION = 1
+
+
+def _copy_json_mapping(value: Optional[dict[str, JSONValue]]) -> Optional[dict[str, JSONValue]]:
+    if value is None:
+        return None
+    return deepcopy(value)
+
+
+def _read_json_mapping(payload: dict[str, Any], key: str) -> Optional[dict[str, JSONValue]]:
+    value = payload.get(key)
+    if value is None:
+        return None
+    if not isinstance(value, dict):
+        raise ValueError(f"整理计划字段 {key} 必须是 JSON 对象")
+    return deepcopy(value)
+
+
+def _read_json_tuple(payload: dict[str, Any], key: str) -> tuple[dict[str, JSONValue], ...]:
+    value = payload.get(key, [])
+    if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
+        raise ValueError(f"整理计划字段 {key} 必须是 JSON 对象数组")
+    return tuple(deepcopy(item) for item in value)
+
+
+def _canonical_json(payload: dict[str, JSONValue]) -> str:
+    try:
+        return json.dumps(
+            payload,
+            ensure_ascii=True,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+    except (TypeError, ValueError) as error:
+        raise ValueError("整理计划只能包含有限 JSON 值") from error
+
+
+@dataclass(frozen=True, slots=True)
+class TransferPlanningInput:
+    """保存可跨重启重放的版本化整理规划输入。"""
+
+    source_fileitem: dict[str, JSONValue]
+    meta: Optional[dict[str, JSONValue]] = None
+    mediainfo: Optional[dict[str, JSONValue]] = None
+    target_directory: Optional[dict[str, JSONValue]] = None
+    target_storage: Optional[str] = None
+    target_path: Optional[str] = None
+    requested_transfer_type: Optional[str] = None
+    media_source: Optional[str] = None
+    media_id: Optional[str] = None
+    media_type: Optional[str] = None
+    need_scrape: bool = False
+    need_rename: bool = True
+    need_notify: bool = True
+    overwrite_mode: Optional[str] = None
+    episodes_info: tuple[dict[str, JSONValue], ...] = ()
+    preview: bool = False
+    options: dict[str, JSONValue] = field(default_factory=dict)
+    schema_version: int = TRANSFER_PLANNING_INPUT_VERSION
+
+    def __post_init__(self) -> None:
+        """拒绝不可恢复的输入版本或缺少源文件身份的快照。"""
+        if self.schema_version != TRANSFER_PLANNING_INPUT_VERSION:
+            raise ValueError(f"不支持的整理规划输入版本: {self.schema_version}")
+        object.__setattr__(self, "source_fileitem", deepcopy(self.source_fileitem))
+        object.__setattr__(self, "meta", _copy_json_mapping(self.meta))
+        object.__setattr__(self, "mediainfo", _copy_json_mapping(self.mediainfo))
+        object.__setattr__(
+            self,
+            "target_directory",
+            _copy_json_mapping(self.target_directory),
+        )
+        object.__setattr__(
+            self,
+            "episodes_info",
+            tuple(deepcopy(item) for item in self.episodes_info),
+        )
+        object.__setattr__(self, "options", deepcopy(self.options))
+        storage = self.source_fileitem.get("storage")
+        path = self.source_fileitem.get("path")
+        if not isinstance(storage, str) or not storage or not isinstance(path, str) or not path:
+            raise ValueError("整理规划输入缺少源文件存储或路径")
+        _canonical_json(self.to_payload())
+
+    @classmethod
+    def legacy(cls, *, storage: str, src_path: str) -> "TransferPlanningInput":
+        """为升级前只有存储与路径的登记构造保守重规划输入。"""
+        return cls(
+            source_fileitem={"storage": storage, "path": src_path},
+            options={"legacy_replan": True},
+        )
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "TransferPlanningInput":
+        """从受版本约束的 JSON 对象恢复规划输入。"""
+        if not isinstance(payload, dict):
+            raise ValueError("整理规划输入必须是 JSON 对象")
+        source_fileitem = _read_json_mapping(payload, "source_fileitem")
+        if source_fileitem is None:
+            raise ValueError("整理规划输入缺少 source_fileitem")
+        return cls(
+            source_fileitem=source_fileitem,
+            meta=_read_json_mapping(payload, "meta"),
+            mediainfo=_read_json_mapping(payload, "mediainfo"),
+            target_directory=_read_json_mapping(payload, "target_directory"),
+            target_storage=payload.get("target_storage"),
+            target_path=payload.get("target_path"),
+            requested_transfer_type=payload.get("requested_transfer_type"),
+            media_source=payload.get("media_source"),
+            media_id=payload.get("media_id"),
+            media_type=payload.get("media_type"),
+            need_scrape=payload.get("need_scrape", False),
+            need_rename=payload.get("need_rename", True),
+            need_notify=payload.get("need_notify", True),
+            overwrite_mode=payload.get("overwrite_mode"),
+            episodes_info=_read_json_tuple(payload, "episodes_info"),
+            preview=payload.get("preview", False),
+            options=_read_json_mapping(payload, "options") or {},
+            schema_version=payload.get("schema_version", 0),
+        )
+
+    def to_payload(self) -> dict[str, JSONValue]:
+        """生成仅含 JSON 值且字段稳定的规划输入投影。"""
+        return {
+            "schema_version": self.schema_version,
+            "source_fileitem": deepcopy(self.source_fileitem),
+            "meta": _copy_json_mapping(self.meta),
+            "mediainfo": _copy_json_mapping(self.mediainfo),
+            "target_directory": _copy_json_mapping(self.target_directory),
+            "target_storage": self.target_storage,
+            "target_path": self.target_path,
+            "requested_transfer_type": self.requested_transfer_type,
+            "media_source": self.media_source,
+            "media_id": self.media_id,
+            "media_type": self.media_type,
+            "need_scrape": self.need_scrape,
+            "need_rename": self.need_rename,
+            "need_notify": self.need_notify,
+            "overwrite_mode": self.overwrite_mode,
+            "episodes_info": [deepcopy(item) for item in self.episodes_info],
+            "preview": self.preview,
+            "options": deepcopy(self.options),
+        }
+
+    @property
+    def fingerprint(self) -> str:
+        """返回规范 JSON 的稳定 SHA-256 指纹。"""
+        return hashlib.sha256(_canonical_json(self.to_payload()).encode("utf-8")).hexdigest()
+
+
+@dataclass(frozen=True, slots=True)
+class TransferPlanItem:
+    """描述规划后按序执行的一条叶子文件操作。"""
+
+    sequence: int
+    source_fileitem: dict[str, JSONValue]
+    target_storage: str
+    target_path: str
+    action: str = "transfer"
+
+    def __post_init__(self) -> None:
+        """拒绝无法定位源文件或目标文件的计划项。"""
+        object.__setattr__(self, "source_fileitem", deepcopy(self.source_fileitem))
+        if self.sequence < 0:
+            raise ValueError("整理计划项序号不能小于零")
+        if not self.target_storage or not self.target_path or not self.action:
+            raise ValueError("整理计划项缺少目标身份或动作")
+        if not self.source_fileitem.get("storage") or not self.source_fileitem.get("path"):
+            raise ValueError("整理计划项缺少源文件身份")
+        _canonical_json(self.to_payload())
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "TransferPlanItem":
+        """从 JSON 对象恢复单条叶子文件操作。"""
+        source_fileitem = _read_json_mapping(payload, "source_fileitem")
+        if source_fileitem is None:
+            raise ValueError("整理计划项缺少 source_fileitem")
+        return cls(
+            sequence=payload.get("sequence", -1),
+            source_fileitem=source_fileitem,
+            target_storage=payload.get("target_storage", ""),
+            target_path=payload.get("target_path", ""),
+            action=payload.get("action", "transfer"),
+        )
+
+    def to_payload(self) -> dict[str, JSONValue]:
+        """生成单条叶子文件操作的 JSON 投影。"""
+        return {
+            "sequence": self.sequence,
+            "source_fileitem": deepcopy(self.source_fileitem),
+            "target_storage": self.target_storage,
+            "target_path": self.target_path,
+            "action": self.action,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class TransferProviderReference:
+    """保存无需依赖运行时 dispatcher 即可持久化的旧 transfer provider 引用。"""
+
+    plugin_id: str
+    plugin_name: str
+    method: str = "transfer"
+
+    def __post_init__(self) -> None:
+        """拒绝无法稳定定位插件或指向非 transfer 方法的引用。"""
+        if not isinstance(self.plugin_id, str) or not self.plugin_id.strip():
+            raise ValueError("旧 transfer provider 的 plugin_id 必须是非空字符串")
+        if not isinstance(self.plugin_name, str) or not self.plugin_name.strip():
+            raise ValueError("旧 transfer provider 的 plugin_name 必须是非空字符串")
+        if self.method != "transfer":
+            raise ValueError("旧 transfer provider 的 method 必须是 transfer")
+        _canonical_json(self.to_payload())
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "TransferProviderReference":
+        """从 JSON 对象恢复旧 transfer provider 引用。"""
+        if not isinstance(payload, dict):
+            raise ValueError("旧 transfer provider 引用必须是 JSON 对象")
+        return cls(
+            plugin_id=payload.get("plugin_id", ""),
+            plugin_name=payload.get("plugin_name", ""),
+            method=payload.get("method", "transfer"),
+        )
+
+    def to_payload(self) -> dict[str, JSONValue]:
+        """生成仅包含稳定插件身份与方法名的 JSON 投影。"""
+        return {
+            "plugin_id": self.plugin_id,
+            "plugin_name": self.plugin_name,
+            "method": self.method,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class TransferProviderInvocationSnapshot:
+    """冻结旧 transfer ABI 中可跨重启恢复的精确参数值。"""
+
+    fileitem: dict[str, JSONValue]
+    meta: Optional[dict[str, JSONValue]]
+    meta_kind: Optional[str]
+    mediainfo: Optional[dict[str, JSONValue]]
+    mediainfo_kind: Optional[str]
+    target_directory: Optional[dict[str, JSONValue]] = None
+    target_storage: Optional[str] = None
+    target_path: Optional[str] = None
+    transfer_type: Optional[str] = None
+    scrape: Optional[bool] = None
+    library_type_folder: Optional[bool] = None
+    library_category_folder: Optional[bool] = None
+    episodes_info: tuple[dict[str, JSONValue], ...] = ()
+    preview: bool = False
+    schema_version: int = TRANSFER_PROVIDER_INVOCATION_VERSION
+
+    def __post_init__(self) -> None:
+        """拒绝缺少源身份、类型不稳定或版本未知的 provider 快照。"""
+        if self.schema_version != TRANSFER_PROVIDER_INVOCATION_VERSION:
+            raise ValueError(
+                f"不支持的旧 transfer provider 调用快照版本: {self.schema_version}"
+            )
+        object.__setattr__(self, "fileitem", deepcopy(self.fileitem))
+        object.__setattr__(self, "meta", _copy_json_mapping(self.meta))
+        object.__setattr__(self, "mediainfo", _copy_json_mapping(self.mediainfo))
+        object.__setattr__(
+            self,
+            "target_directory",
+            _copy_json_mapping(self.target_directory),
+        )
+        object.__setattr__(
+            self,
+            "episodes_info",
+            tuple(deepcopy(item) for item in self.episodes_info),
+        )
+        if not self.fileitem.get("storage") or not self.fileitem.get("path"):
+            raise ValueError("旧 transfer provider 调用快照缺少源文件身份")
+        for kind in (self.meta_kind, self.mediainfo_kind):
+            if kind is not None and (not isinstance(kind, str) or not kind):
+                raise ValueError("旧 transfer provider 调用快照类型必须是非空字符串")
+        for value in (
+                self.target_storage,
+                self.target_path,
+                self.transfer_type,
+        ):
+            if value is not None and not isinstance(value, str):
+                raise ValueError("旧 transfer provider 可选文本参数必须是字符串或 None")
+        for bool_value in (
+                self.scrape,
+                self.library_type_folder,
+                self.library_category_folder,
+        ):
+            if bool_value is not None and not isinstance(bool_value, bool):
+                raise ValueError("旧 transfer provider 可选布尔参数必须是 bool 或 None")
+        if not isinstance(self.preview, bool):
+            raise ValueError("旧 transfer provider preview 参数必须是 bool")
+        _canonical_json(self.to_payload())
+
+    @classmethod
+    def from_payload(
+            cls,
+            payload: dict[str, Any],
+    ) -> "TransferProviderInvocationSnapshot":
+        """从版本化 JSON 对象严格恢复旧 transfer ABI 调用快照。"""
+        if not isinstance(payload, dict):
+            raise ValueError("旧 transfer provider 调用快照必须是 JSON 对象")
+        fileitem = _read_json_mapping(payload, "fileitem")
+        if fileitem is None:
+            raise ValueError("旧 transfer provider 调用快照缺少 fileitem")
+        return cls(
+            fileitem=fileitem,
+            meta=_read_json_mapping(payload, "meta"),
+            meta_kind=payload.get("meta_kind"),
+            mediainfo=_read_json_mapping(payload, "mediainfo"),
+            mediainfo_kind=payload.get("mediainfo_kind"),
+            target_directory=_read_json_mapping(payload, "target_directory"),
+            target_storage=payload.get("target_storage"),
+            target_path=payload.get("target_path"),
+            transfer_type=payload.get("transfer_type"),
+            scrape=payload.get("scrape"),
+            library_type_folder=payload.get("library_type_folder"),
+            library_category_folder=payload.get("library_category_folder"),
+            episodes_info=_read_json_tuple(payload, "episodes_info"),
+            preview=payload.get("preview", False),
+            schema_version=payload.get("schema_version", 0),
+        )
+
+    def to_payload(self) -> dict[str, JSONValue]:
+        """生成仅含 JSON 值且保留 None 与 False 差异的调用投影。"""
+        return {
+            "schema_version": self.schema_version,
+            "fileitem": deepcopy(self.fileitem),
+            "meta": _copy_json_mapping(self.meta),
+            "meta_kind": self.meta_kind,
+            "mediainfo": _copy_json_mapping(self.mediainfo),
+            "mediainfo_kind": self.mediainfo_kind,
+            "target_directory": _copy_json_mapping(self.target_directory),
+            "target_storage": self.target_storage,
+            "target_path": self.target_path,
+            "transfer_type": self.transfer_type,
+            "scrape": self.scrape,
+            "library_type_folder": self.library_type_folder,
+            "library_category_folder": self.library_category_folder,
+            "episodes_info": [deepcopy(item) for item in self.episodes_info],
+            "preview": self.preview,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class TransferPlanCheckpoint:
+    """保存无需重触发识别或重命名即可执行的完整有序计划。"""
+
+    planning_input: TransferPlanningInput
+    target_storage: str
+    root_target_path: str
+    final_target_path: str
+    resolved_transfer_type: str
+    items: tuple[TransferPlanItem, ...]
+    resolved_meta: Optional[dict[str, JSONValue]] = None
+    resolved_meta_kind: Optional[str] = None
+    resolved_mediainfo: Optional[dict[str, JSONValue]] = None
+    resolved_mediainfo_kind: Optional[str] = None
+    resolved_episodes_info: tuple[dict[str, JSONValue], ...] = ()
+    legacy_transfer_providers: tuple[TransferProviderReference, ...] = ()
+    provider_invocation: Optional[TransferProviderInvocationSnapshot] = None
+    pre_execution_cleanup_completed: bool = False
+    need_scrape: bool = False
+    need_rename: bool = False
+    need_notify: bool = True
+    overwrite_mode: Optional[str] = None
+    preview: bool = False
+    skip_reason: Optional[str] = None
+    schema_version: int = TRANSFER_PLAN_CHECKPOINT_VERSION
+
+    def __post_init__(self) -> None:
+        """验证版本、目标身份和计划项顺序组成完整检查点。"""
+        if self.schema_version != TRANSFER_PLAN_CHECKPOINT_VERSION:
+            raise ValueError(f"不支持的整理计划检查点版本: {self.schema_version}")
+        if not isinstance(self.pre_execution_cleanup_completed, bool):
+            raise ValueError("整理计划预执行 cleanup 完成标记必须是 bool")
+        object.__setattr__(self, "resolved_meta", _copy_json_mapping(self.resolved_meta))
+        object.__setattr__(
+            self,
+            "resolved_mediainfo",
+            _copy_json_mapping(self.resolved_mediainfo),
+        )
+        object.__setattr__(
+            self,
+            "resolved_episodes_info",
+            tuple(deepcopy(item) for item in self.resolved_episodes_info),
+        )
+        if not isinstance(self.legacy_transfer_providers, tuple) or any(
+                not isinstance(provider, TransferProviderReference)
+                for provider in self.legacy_transfer_providers
+        ):
+            raise ValueError("整理计划的旧 transfer provider 引用必须是类型化元组")
+        provider_ids = tuple(
+            provider.plugin_id for provider in self.legacy_transfer_providers
+        )
+        if len(set(provider_ids)) != len(provider_ids):
+            raise ValueError("整理计划的旧 transfer provider plugin_id 不能重复")
+        for resolved_kind in (
+                self.resolved_meta_kind,
+                self.resolved_mediainfo_kind,
+        ):
+            if resolved_kind is not None and (
+                    not isinstance(resolved_kind, str) or not resolved_kind
+            ):
+                raise ValueError("整理计划的已解析上下文类型必须是非空字符串")
+        if self.provider_invocation is not None:
+            if not self.legacy_transfer_providers:
+                raise ValueError("provider_pending 检查点缺少冻结 provider")
+            if self.pre_execution_cleanup_completed:
+                raise ValueError("provider_pending 提交时 cleanup 尚未执行")
+            if (
+                    not self.provider_invocation.meta
+                    or not self.provider_invocation.meta_kind
+                    or not self.provider_invocation.mediainfo
+                    or not self.provider_invocation.mediainfo_kind
+            ):
+                raise ValueError("provider_pending 检查点缺少可重放的媒体上下文")
+            if (
+                    self.target_storage
+                    or self.root_target_path
+                    or self.final_target_path
+                    or self.resolved_transfer_type
+                    or self.items
+                    or self.skip_reason
+            ):
+                raise ValueError("provider_pending 检查点不得包含宿主执行计划")
+        else:
+            if not self.target_storage or not self.root_target_path or not self.final_target_path:
+                raise ValueError("整理计划检查点缺少目标身份")
+            if not self.resolved_transfer_type:
+                raise ValueError("整理计划检查点缺少已解析的整理方式")
+        if tuple(item.sequence for item in self.items) != tuple(range(len(self.items))):
+            raise ValueError("整理计划项必须按从零开始的连续序号保存")
+        if (
+                self.provider_invocation is None
+                and not self.items
+                and not self.preview
+                and not self.skip_reason
+        ):
+            raise ValueError("非预览空计划必须记录合法跳过原因")
+        _canonical_json(self.to_payload())
+
+    @property
+    def is_provider_pending(self) -> bool:
+        """返回该检查点是否只冻结 provider 调用、尚未生成宿主计划。"""
+        return self.provider_invocation is not None
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "TransferPlanCheckpoint":
+        """从受版本约束的 JSON 对象恢复完整执行检查点。"""
+        if not isinstance(payload, dict):
+            raise ValueError("整理计划检查点必须是 JSON 对象")
+        planning_input_payload = _read_json_mapping(payload, "planning_input")
+        if planning_input_payload is None:
+            raise ValueError("整理计划检查点缺少 planning_input")
+        item_payloads = payload.get("items", [])
+        if not isinstance(item_payloads, list) or not all(
+                isinstance(item, dict) for item in item_payloads
+        ):
+            raise ValueError("整理计划检查点 items 必须是 JSON 对象数组")
+        legacy_provider_payloads = payload.get("legacy_transfer_providers", [])
+        if not isinstance(legacy_provider_payloads, list):
+            raise ValueError(
+                "整理计划检查点 legacy_transfer_providers 必须是 JSON 对象数组"
+            )
+        provider_invocation_payload = payload.get("provider_invocation")
+        if provider_invocation_payload is not None and not isinstance(
+                provider_invocation_payload,
+                dict,
+        ):
+            raise ValueError("整理计划检查点 provider_invocation 必须是 JSON 对象")
+        return cls(
+            planning_input=TransferPlanningInput.from_payload(planning_input_payload),
+            target_storage=payload.get("target_storage", ""),
+            root_target_path=payload.get("root_target_path", ""),
+            final_target_path=payload.get("final_target_path", ""),
+            resolved_transfer_type=payload.get("resolved_transfer_type", ""),
+            items=tuple(TransferPlanItem.from_payload(item) for item in item_payloads),
+            resolved_meta=_read_json_mapping(payload, "resolved_meta"),
+            resolved_meta_kind=payload.get("resolved_meta_kind"),
+            resolved_mediainfo=_read_json_mapping(payload, "resolved_mediainfo"),
+            resolved_mediainfo_kind=payload.get("resolved_mediainfo_kind"),
+            resolved_episodes_info=_read_json_tuple(
+                payload,
+                "resolved_episodes_info",
+            ),
+            legacy_transfer_providers=tuple(
+                TransferProviderReference.from_payload(provider)
+                for provider in legacy_provider_payloads
+            ),
+            provider_invocation=(
+                TransferProviderInvocationSnapshot.from_payload(
+                    provider_invocation_payload
+                )
+                if provider_invocation_payload is not None
+                else None
+            ),
+            pre_execution_cleanup_completed=payload.get(
+                "pre_execution_cleanup_completed",
+                False,
+            ),
+            need_scrape=payload.get("need_scrape", False),
+            need_rename=payload.get("need_rename", False),
+            need_notify=payload.get("need_notify", True),
+            overwrite_mode=payload.get("overwrite_mode"),
+            preview=payload.get("preview", False),
+            skip_reason=payload.get("skip_reason"),
+            schema_version=payload.get("schema_version", 0),
+        )
+
+    def to_payload(self) -> dict[str, JSONValue]:
+        """生成可原子落库的完整版本化 JSON 检查点。"""
+        return {
+            "schema_version": self.schema_version,
+            "planning_input": self.planning_input.to_payload(),
+            "target_storage": self.target_storage,
+            "root_target_path": self.root_target_path,
+            "final_target_path": self.final_target_path,
+            "resolved_transfer_type": self.resolved_transfer_type,
+            "items": [item.to_payload() for item in self.items],
+            "resolved_meta": _copy_json_mapping(self.resolved_meta),
+            "resolved_meta_kind": self.resolved_meta_kind,
+            "resolved_mediainfo": _copy_json_mapping(self.resolved_mediainfo),
+            "resolved_mediainfo_kind": self.resolved_mediainfo_kind,
+            "resolved_episodes_info": [
+                deepcopy(item) for item in self.resolved_episodes_info
+            ],
+            "legacy_transfer_providers": [
+                provider.to_payload()
+                for provider in self.legacy_transfer_providers
+            ],
+            "provider_invocation": (
+                self.provider_invocation.to_payload()
+                if self.provider_invocation
+                else None
+            ),
+            "pre_execution_cleanup_completed": (
+                self.pre_execution_cleanup_completed
+            ),
+            "need_scrape": self.need_scrape,
+            "need_rename": self.need_rename,
+            "need_notify": self.need_notify,
+            "overwrite_mode": self.overwrite_mode,
+            "preview": self.preview,
+            "skip_reason": self.skip_reason,
+        }
+
+
+class TransferAdmissionConflictError(ValueError):
+    """同一源文件以不同规划输入重复准入时抛出的冲突错误。"""
+
+
+class TransferPlanningStateError(RuntimeError):
+    """计划检查点无法从当前持久状态推进时抛出的状态错误。"""
+
 
 class TransferTask(OptionalMediaIdentityMixin, BaseModel):
     """
@@ -81,6 +658,9 @@ class TransferTask(OptionalMediaIdentityMixin, BaseModel):
     background: Optional[bool] = True
     preview: Optional[bool] = False
     _admission_task_id: Optional[str] = PrivateAttr(default=None)
+    _planning_input: Optional[TransferPlanningInput] = PrivateAttr(default=None)
+    _plan_checkpoint: Optional[TransferPlanCheckpoint] = PrivateAttr(default=None)
+    _planning_context_restored: bool = PrivateAttr(default=False)
 
     @property
     def admission_task_id(self) -> Optional[str]:
@@ -91,6 +671,33 @@ class TransferTask(OptionalMediaIdentityMixin, BaseModel):
         """绑定持久准入生成的稳定身份，不改变插件可见序列化字段。"""
         self._admission_task_id = task_id
 
+    @property
+    def planning_input(self) -> Optional[TransferPlanningInput]:
+        """返回宿主恢复规划使用的内部输入快照。"""
+        return self._planning_input
+
+    @property
+    def plan_checkpoint(self) -> Optional[TransferPlanCheckpoint]:
+        """返回宿主直接执行已规划任务使用的内部检查点。"""
+        return self._plan_checkpoint
+
+    def bind_planning_input(self, planning_input: TransferPlanningInput) -> None:
+        """绑定持久规划输入，不改变插件可见序列化字段。"""
+        self._planning_input = planning_input
+
+    def bind_plan_checkpoint(self, checkpoint: TransferPlanCheckpoint) -> None:
+        """绑定持久执行检查点，不改变插件可见序列化字段。"""
+        self._plan_checkpoint = checkpoint
+
+    @property
+    def planning_context_restored(self) -> bool:
+        """返回当前领域上下文是否来自持久快照。"""
+        return self._planning_context_restored
+
+    def mark_planning_context_restored(self) -> None:
+        """标记领域上下文已离线恢复，禁止旧流程再次在线补充。"""
+        self._planning_context_restored = True
+
     def to_dict(self):
         """
         返回字典。
@@ -122,9 +729,6 @@ class TransferQueue(BaseModel):
     result: Optional[TransferInfo] = None
 
 
-TRANSFER_ADMISSION_ACCEPTED = "accepted"
-
-
 @dataclass(frozen=True, slots=True)
 class TransferAdmission:
     """描述已经持久化、可在进程退出后恢复的整理任务准入事实。"""
@@ -136,13 +740,22 @@ class TransferAdmission:
     created_at: str
     updated_at: str
     last_error: Optional[str] = None
+    input_fingerprint: Optional[str] = None
+    planning_input: Optional[TransferPlanningInput] = None
+    checkpoint: Optional[TransferPlanCheckpoint] = None
 
 
 class TransferAdmissionRepository(Protocol):
     """整理任务 durable admission 所需的类型化持久化端口。"""
 
-    def admit(self, *, storage: str, src_path: str) -> TransferAdmission:
-        """幂等登记源文件并返回稳定任务身份。"""
+    def admit(
+            self,
+            *,
+            storage: str,
+            src_path: str,
+            planning_input: Optional[TransferPlanningInput] = None,
+    ) -> TransferAdmission:
+        """按规划输入幂等登记源文件并返回稳定任务身份。"""
         ...
 
     def list_accepted(self, limit: int = 5000) -> list[TransferAdmission]:
@@ -153,6 +766,24 @@ class TransferAdmissionRepository(Protocol):
         """记录内存队列接收失败，保留任务供后续恢复。"""
         ...
 
+    def checkpoint_plan(
+            self,
+            *,
+            task_id: str,
+            input_fingerprint: str,
+            checkpoint: TransferPlanCheckpoint,
+    ) -> TransferAdmission:
+        """原子保存完整计划并将匹配输入的任务推进到已规划。"""
+        ...
+
+    def record_planning_failure(self, *, task_id: str, error: str) -> None:
+        """记录规划失败但保留接纳状态供下次恢复重试。"""
+        ...
+
+    def list_recoverable(self, limit: int = 5000) -> list[TransferAdmission]:
+        """按登记顺序返回接纳或已规划的可恢复任务。"""
+        ...
+
     def discard_task(self, *, task_id: str) -> int:
         """按稳定任务身份删除已经到达终态的登记。"""
         ...

app/chain/__init__.py

@@ -5,7 +5,7 @@ import traceback
 from abc import ABCMeta
 from collections.abc import Callable
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union, cast
 
 from app.application.chain.context import ChainRuntimeContext, get_chain_runtime_context
 from app.application.chain.data import get_chain_data_ports
@@ -15,7 +15,7 @@ from app.application.configuration import (
 )
 from app.chain._messaging import MessageProcessingMixin, NotificationMixin
 from app.chain._recognition import RecognitionMixin
-from app.domain.context import Context, MediaInfo, SubtitleInfo, TorrentInfo
+from app.domain.context import Context, MediaInfo, MusicInfo, SubtitleInfo, TorrentInfo
 from app.domain.meta.metabase import MetaBase
 from app.runtime.log import logger
 from app.schemas.category import CategoryConfig
@@ -35,6 +35,9 @@ from app.schemas.types import (
 )
 from app.schemas.workflow import FileItem
 
+if TYPE_CHECKING:
+    from app.application.transfer import TransferPlanCheckpoint, TransferPlanningInput
+
 
 class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
                 metaclass=ABCMeta):
@@ -65,6 +68,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
             system_error_handler=self.__handle_system_error,
             rate_limit_handler=self.__handle_rate_limit_error,
         )
+        self._legacy_transfer_command = context.legacy_transfer_command
         self.messagequeue = context.message_queue_factory(self.run_module)
 
     @property
@@ -216,6 +220,15 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
         """
         return self._module_dispatcher.dispatch(method, *args, **kwargs)
 
+    def run_module_strict(
+            self,
+            method: str,
+            *args: Any,
+            **kwargs: Any,
+    ) -> Any:
+        """运行模块并传播 provider 异常，供必须区分空结果与查询失败的能力使用。"""
+        return self._module_dispatcher.dispatch_strict(method, *args, **kwargs)
+
     async def async_run_module(
             self,
             method: str,
@@ -854,7 +867,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
             self,
             fileitem: FileItem,
             meta: MetaBase,
-            mediainfo: MediaInfo,
+            mediainfo: Union[MediaInfo, MusicInfo],
             target_directory: TransferDirectoryConf = None,
             target_storage: Optional[str] = None,
             target_path: Path = None,
@@ -868,7 +881,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
             preview: bool = False,
     ) -> Optional[TransferInfo]:
         """
-        文件转移
+        经启动组合根注入的 canonical durable command 执行旧整理 ABI。
         :param fileitem:  文件信息
         :param meta: 预识别的元数据
         :param mediainfo:  识别的媒体信息
@@ -885,8 +898,9 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
         :param preview: 是否仅预览，不执行实际转移
         :return: {path, target_path, message}
         """
-        return self.run_module(
-            "transfer",
+        if self._legacy_transfer_command is None:
+            raise RuntimeError("旧整理兼容命令尚未由启动组合根配置")
+        return self._legacy_transfer_command(
             fileitem=fileitem,
             meta=meta,
             mediainfo=mediainfo,
@@ -903,6 +917,69 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
             preview=preview,
         )
 
+    def plan_transfer(
+            self,
+            fileitem: FileItem,
+            meta: MetaBase,
+            mediainfo: Union[MediaInfo, MusicInfo],
+            target_directory: Optional[TransferDirectoryConf] = None,
+            target_storage: Optional[str] = None,
+            target_path: Optional[Path] = None,
+            transfer_type: Optional[str] = None,
+            scrape: Optional[bool] = None,
+            library_type_folder: Optional[bool] = None,
+            library_category_folder: Optional[bool] = None,
+            episodes_info: Optional[List[TmdbEpisode]] = None,
+            source_oper: Any = None,
+            preview: bool = False,
+            planning_input: Optional[TransferPlanningInput] = None,
+    ) -> Optional[TransferPlanCheckpoint]:
+        """调用文件管理模块生成无文件写入的冻结整理计划。"""
+        return cast(
+            "Optional[TransferPlanCheckpoint]",
+            self.run_module(
+                "plan_transfer",
+                fileitem=fileitem,
+                meta=meta,
+                mediainfo=mediainfo,
+                target_directory=target_directory,
+                target_path=target_path,
+                target_storage=target_storage,
+                transfer_type=transfer_type,
+                scrape=scrape,
+                library_type_folder=library_type_folder,
+                library_category_folder=library_category_folder,
+                episodes_info=episodes_info,
+                source_oper=source_oper,
+                preview=preview,
+                planning_input=planning_input,
+            ),
+        )
+
+    def execute_transfer_plan(
+            self,
+            checkpoint: TransferPlanCheckpoint,
+            *,
+            meta: MetaBase,
+            mediainfo: Union[MediaInfo, MusicInfo],
+            source_oper: Any = None,
+            target_oper: Any = None,
+            cleanup_media_file: Optional[Callable[[FileItem], bool]] = None,
+    ) -> Optional[TransferInfo]:
+        """调用文件管理模块执行已冻结计划，并注入统一安全删除能力。"""
+        return cast(
+            Optional[TransferInfo],
+            self.run_module(
+                "execute_transfer_plan",
+                checkpoint=checkpoint,
+                meta=meta,
+                mediainfo=mediainfo,
+                source_oper=source_oper,
+                target_oper=target_oper,
+                cleanup_media_file=cleanup_media_file,
+            ),
+        )
+
     def transfer_completed(self, hashs: str, downloader: Optional[str] = None) -> None:
         """
         下载器转移完成后的处理

app/chain/storage.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, cast
 
 from app.application.directory import DirectoryHelper
 from app.chain import ChainBase
@@ -99,6 +99,21 @@ class StorageChain(ChainBase):
         """
         return self.run_module("get_file_item", storage=storage, path=path)
 
+    def get_file_item_strict(
+            self,
+            storage: str,
+            path: Path,
+    ) -> Optional[_SchemaFileItem]:
+        """严格查询文件项：确认不存在返回空，provider 或 I/O 失败直接抛出。"""
+        return cast(
+            Optional[_SchemaFileItem],
+            self.run_module_strict(
+                "get_file_item",
+                storage=storage,
+                path=path,
+            ),
+        )
+
     def get_parent_item(self, fileitem: _SchemaFileItem) -> Optional[_SchemaFileItem]:
         """
         获取上级目录项

app/db/adapters/transfer.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Callable
 from datetime import datetime
+from typing import Optional
 from uuid import uuid4
 
 from sqlalchemy.exc import IntegrityError
@@ -9,7 +10,13 @@ from sqlalchemy.orm import Session
 
 from app.application.transfer import (
     TRANSFER_ADMISSION_ACCEPTED,
+    TRANSFER_ADMISSION_PLANNED,
+    TRANSFER_ADMISSION_PROVIDER_PENDING,
     TransferAdmission,
+    TransferAdmissionConflictError,
+    TransferPlanCheckpoint,
+    TransferPlanningInput,
+    TransferPlanningStateError,
 )
 from app.db.models.transferpending import TransferPending
 from app.db.oper.transferpending import TransferPendingOper
@@ -32,6 +39,40 @@ class TransactionalTransferAdmissionRepository:
     def _project(pending: TransferPending) -> TransferAdmission:
         """在 Session 有效期内把 ORM 行冻结为应用层 DTO。"""
         created_at = pending.created_at or pending.updated_at
+        planning_input = TransferPlanningInput.from_payload(pending.planning_input)
+        if pending.input_version != planning_input.schema_version:
+            raise TransferPlanningStateError("整理规划输入列版本与 JSON 版本不一致")
+        if pending.input_fingerprint != planning_input.fingerprint:
+            raise TransferAdmissionConflictError("整理规划输入 JSON 与持久指纹不一致")
+        checkpoint = (
+            TransferPlanCheckpoint.from_payload(pending.checkpoint_payload)
+            if pending.checkpoint_payload is not None
+            else None
+        )
+        if checkpoint is not None:
+            if pending.checkpoint_version != checkpoint.schema_version:
+                raise TransferPlanningStateError("整理检查点列版本与 JSON 版本不一致")
+            if checkpoint.planning_input.fingerprint != pending.input_fingerprint:
+                raise TransferAdmissionConflictError("整理检查点内嵌输入与准入指纹不一致")
+        if pending.state in {
+                TRANSFER_ADMISSION_PROVIDER_PENDING,
+                TRANSFER_ADMISSION_PLANNED,
+        } and checkpoint is None:
+            raise TransferPlanningStateError("待执行任务缺少完整检查点")
+        if pending.state == TRANSFER_ADMISSION_ACCEPTED and checkpoint is not None:
+            raise TransferPlanningStateError("接纳态任务不能携带计划检查点")
+        if (
+                pending.state == TRANSFER_ADMISSION_PROVIDER_PENDING
+                and checkpoint is not None
+                and not checkpoint.is_provider_pending
+        ):
+            raise TransferPlanningStateError("provider_pending 状态缺少 provider 调用快照")
+        if (
+                pending.state == TRANSFER_ADMISSION_PLANNED
+                and checkpoint is not None
+                and checkpoint.is_provider_pending
+        ):
+            raise TransferPlanningStateError("planned 状态不能携带 provider-only 检查点")
         return TransferAdmission(
             task_id=pending.task_id,
             storage=pending.storage,
@@ -40,10 +81,39 @@ class TransactionalTransferAdmissionRepository:
             created_at=created_at,
             updated_at=pending.updated_at,
             last_error=pending.last_error,
+            input_fingerprint=pending.input_fingerprint,
+            planning_input=planning_input,
+            checkpoint=checkpoint,
         )
 
-    def admit(self, *, storage: str, src_path: str) -> TransferAdmission:
-        """幂等持久化准入事实，并返回跨重启稳定的任务标识。"""
+    @staticmethod
+    def _assert_input_match(
+            pending: TransferPending,
+            planning_input: TransferPlanningInput,
+    ) -> None:
+        """拒绝同一源文件以不同规划输入复用既有任务身份。"""
+        if pending.input_fingerprint != planning_input.fingerprint:
+            raise TransferAdmissionConflictError(
+                f"整理源文件已按不同输入准入: {pending.storage}:{pending.src_path}"
+            )
+
+    def admit(
+            self,
+            *,
+            storage: str,
+            src_path: str,
+            planning_input: Optional[TransferPlanningInput] = None,
+    ) -> TransferAdmission:
+        """按输入指纹幂等持久化准入事实，并返回跨重启稳定身份。"""
+        effective_input = planning_input or TransferPlanningInput.legacy(
+            storage=storage,
+            src_path=src_path,
+        )
+        if (
+                effective_input.source_fileitem.get("storage") != storage
+                or effective_input.source_fileitem.get("path") != src_path
+        ):
+            raise ValueError("整理规划输入的源文件身份与准入参数不一致")
         now_time = self._now()
         try:
             with self._session_factory() as session:
@@ -55,10 +125,14 @@ class TransactionalTransferAdmissionRepository:
                         src_path=src_path,
                         state=TRANSFER_ADMISSION_ACCEPTED,
                         now_time=now_time,
+                        input_version=effective_input.schema_version,
+                        planning_input=effective_input.to_payload(),
+                        input_fingerprint=effective_input.fingerprint,
                     )
                     if pending is None:
                         raise ValueError("整理任务的存储与源路径不能为空")
                     session.flush()
+                    self._assert_input_match(pending, effective_input)
                     admission = self._project(pending)
                     transaction.commit()
                     return admission
@@ -74,6 +148,7 @@ class TransactionalTransferAdmissionRepository:
                 )
                 if pending is None:
                     raise RuntimeError("并发准入冲突后未找到已提交记录") from error
+                self._assert_input_match(pending, effective_input)
                 return self._project(pending)
 
     def list_accepted(self, limit: int = 5000) -> list[TransferAdmission]:
@@ -85,6 +160,19 @@ class TransactionalTransferAdmissionRepository:
             )
             return [self._project(pending) for pending in pending_items]
 
+    def list_recoverable(self, limit: int = 5000) -> list[TransferAdmission]:
+        """投影接纳、provider 待执行或已规划的全部可恢复任务。"""
+        with self._session_factory() as session:
+            pending_items = TransferPendingOper(db=session).list_by_states(
+                states=(
+                    TRANSFER_ADMISSION_ACCEPTED,
+                    TRANSFER_ADMISSION_PROVIDER_PENDING,
+                    TRANSFER_ADMISSION_PLANNED,
+                ),
+                limit=limit,
+            )
+            return [self._project(pending) for pending in pending_items]
+
     def record_enqueue_failure(self, *, task_id: str, error: str) -> None:
         """独立提交最近一次入队失败，保留准入记录供后续恢复。"""
         with self._session_factory() as session:
@@ -100,6 +188,79 @@ class TransactionalTransferAdmissionRepository:
                 transaction.rollback()
                 raise
 
+    def checkpoint_plan(
+            self,
+            *,
+            task_id: str,
+            input_fingerprint: str,
+            checkpoint: TransferPlanCheckpoint,
+    ) -> TransferAdmission:
+        """以输入指纹 CAS 保存 provider 调用快照或升级宿主计划。"""
+        if checkpoint.planning_input.fingerprint != input_fingerprint:
+            raise TransferAdmissionConflictError("检查点输入与准入输入指纹不一致")
+        checkpoint_payload = checkpoint.to_payload()
+        target_state = (
+            TRANSFER_ADMISSION_PROVIDER_PENDING
+            if checkpoint.is_provider_pending
+            else TRANSFER_ADMISSION_PLANNED
+        )
+        source_states = (
+            (TRANSFER_ADMISSION_ACCEPTED,)
+            if checkpoint.is_provider_pending
+            else (
+                TRANSFER_ADMISSION_ACCEPTED,
+                TRANSFER_ADMISSION_PROVIDER_PENDING,
+            )
+        )
+        with self._session_factory() as session:
+            transaction = SqlAlchemyUnitOfWork(session)
+            try:
+                oper = TransferPendingOper(db=session)
+                updated = oper.stage_checkpoint_plan(
+                    task_id=task_id,
+                    input_fingerprint=input_fingerprint,
+                    checkpoint_version=checkpoint.schema_version,
+                    checkpoint_payload=checkpoint_payload,
+                    source_states=source_states,
+                    target_state=target_state,
+                    now_time=self._now(),
+                )
+                session.flush()
+                session.expire_all()
+                pending = oper.get_by_task_id(task_id=task_id)
+                if pending is None:
+                    raise TransferPlanningStateError(f"未找到整理任务: {task_id}")
+                if pending.input_fingerprint != input_fingerprint:
+                    raise TransferAdmissionConflictError("整理任务输入指纹已经改变")
+                if not updated and not (
+                        pending.state == target_state
+                        and pending.checkpoint_payload == checkpoint_payload
+                ):
+                    raise TransferPlanningStateError(
+                        f"整理任务不能从状态 {pending.state} 保存 {target_state} 检查点"
+                    )
+                admission = self._project(pending)
+                transaction.commit()
+                return admission
+            except Exception:
+                transaction.rollback()
+                raise
+
+    def record_planning_failure(self, *, task_id: str, error: str) -> None:
+        """独立提交规划错误并保持任务处于接纳态供恢复重试。"""
+        with self._session_factory() as session:
+            transaction = SqlAlchemyUnitOfWork(session)
+            try:
+                TransferPendingOper(db=session).stage_record_planning_failure(
+                    task_id=task_id,
+                    error=error,
+                    now_time=self._now(),
+                )
+                transaction.commit()
+            except Exception:
+                transaction.rollback()
+                raise
+
     def discard_task(self, *, task_id: str) -> int:
         """在独立事务中按稳定任务标识删除已到终态的准入记录。"""
         with self._session_factory() as session:

app/db/models/transferpending.py

@@ -1,13 +1,66 @@
+import hashlib
+import json
 from datetime import datetime
-from typing import List, Optional, cast
+from typing import Any, List, Optional, cast
 from uuid import uuid4
 
-from sqlalchemy import Index, String, Text, UniqueConstraint, delete, select, update
+from sqlalchemy import JSON, Index, Integer, String, Text, UniqueConstraint, delete, select, update
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.db.base import Base, execute_dml, get_id_column
 
 
+def _legacy_planning_payload(storage: str, src_path: str) -> dict[str, Any]:
+    return {
+        "schema_version": 1,
+        "source_fileitem": {"storage": storage, "path": src_path},
+        "meta": None,
+        "mediainfo": None,
+        "target_directory": None,
+        "target_storage": None,
+        "target_path": None,
+        "requested_transfer_type": None,
+        "media_source": None,
+        "media_id": None,
+        "media_type": None,
+        "need_scrape": False,
+        "need_rename": True,
+        "need_notify": True,
+        "overwrite_mode": None,
+        "episodes_info": [],
+        "preview": False,
+        "options": {"legacy_replan": True},
+    }
+
+
+def _planning_fingerprint(payload: dict[str, Any]) -> str:
+    canonical = json.dumps(
+        payload,
+        ensure_ascii=True,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    )
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def _default_planning_payload(context: Any) -> dict[str, Any]:
+    params = context.get_current_parameters()
+    return _legacy_planning_payload(
+        params.get("storage", ""),
+        params.get("src_path", ""),
+    )
+
+
+def _default_planning_fingerprint(context: Any) -> str:
+    params = context.get_current_parameters()
+    payload = params.get("planning_input") or _legacy_planning_payload(
+        params.get("storage", ""),
+        params.get("src_path", ""),
+    )
+    return _planning_fingerprint(payload)
+
+
 class TransferPending(Base):
     """
     待整理文件登记。
@@ -17,9 +70,9 @@ class TransferPending(Base):
     蒸发。而已经稳定落地的文件不会再产生任何监控事件，也不会有新的补偿扫描起点
     ——结果就是永久漏件，只能靠人工比对补整理。
 
-    这里只落盘恢复所需的最小事实：稳定任务身份、存储、源文件路径、准入状态和
-    最近入队错误。重启后重新走一遍整理入口，由整理历史查重挡掉已经完成的，
-    因此不需要序列化 meta/mediainfo 这些重对象，也不存在识别结果陈旧的问题。
+    准入时保存版本化规划输入和指纹；纯规划完成后以同一行原子保存完整有序计划并
+    推进到 planned。重启恢复可直接消费已规划路径，避免再次触发 rename 等插件事件。
+    旧路径登记接口仍生成最小 legacy_replan 输入，供插件兼容调用方继续使用。
     """
 
     id = get_id_column()
@@ -42,6 +95,22 @@ class TransferPending(Base):
     )
     # 最近一次入队失败原因
     last_error: Mapped[Optional[str]] = mapped_column(Text)
+    # 规划输入格式版本
+    input_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    # 版本化规划输入 JSON
+    planning_input: Mapped[dict[str, Any]] = mapped_column(
+        JSON, nullable=False, default=_default_planning_payload
+    )
+    # 规划输入规范 JSON 的 SHA-256 指纹
+    input_fingerprint: Mapped[str] = mapped_column(
+        String(64), nullable=False, default=_default_planning_fingerprint
+    )
+    # 完整计划格式版本，尚未规划时为空
+    checkpoint_version: Mapped[Optional[int]] = mapped_column(Integer)
+    # 完整有序计划 JSON，尚未规划时为空
+    checkpoint_payload: Mapped[Optional[dict[str, Any]]] = mapped_column(JSON)
+    # 规划完成时间
+    planned_at: Mapped[Optional[str]] = mapped_column(String(40))
 
     __table_args__ = (
         # 同一个文件重复入队只保留一条，回放时不会重复送入整理链
@@ -74,12 +143,16 @@ class TransferPending(Base):
         ).scalars().first()
         if pending:
             return cast("TransferPending", pending)
+        planning_input = _legacy_planning_payload(storage, src_path)
         pending = cls(
             storage=storage,
             src_path=src_path,
             state="accepted",
             created_at=now_time,
             updated_at=now_time,
+            input_version=1,
+            planning_input=planning_input,
+            input_fingerprint=_planning_fingerprint(planning_input),
         )
         db.add(pending)
         return pending
@@ -87,7 +160,9 @@ class TransferPending(Base):
     @classmethod
     def stage_admit(cls, db: Session, *, task_id: str, storage: str,
                     src_path: str, state: str,
-                    now_time: str) -> Optional["TransferPending"]:
+                    now_time: str, input_version: int = 1,
+                    planning_input: Optional[dict[str, Any]] = None,
+                    input_fingerprint: Optional[str] = None) -> Optional["TransferPending"]:
         """
         在调用方会话中暂存一条持久接纳记录。
 
@@ -98,6 +173,9 @@ class TransferPending(Base):
         :param src_path: 源文件路径
         :param state: 持久状态
         :param now_time: 当前时间
+        :param input_version: 规划输入格式版本
+        :param planning_input: 版本化规划输入 JSON
+        :param input_fingerprint: 规划输入规范 JSON 指纹
         :return: 接纳记录
         """
         if not task_id or not storage or not src_path or not state:
@@ -107,6 +185,8 @@ class TransferPending(Base):
         ).scalars().first()
         if pending:
             return cast("TransferPending", pending)
+        effective_input = planning_input or _legacy_planning_payload(storage, src_path)
+        effective_fingerprint = input_fingerprint or _planning_fingerprint(effective_input)
         pending = cls(
             task_id=task_id,
             storage=storage,
@@ -114,6 +194,9 @@ class TransferPending(Base):
             state=state,
             created_at=now_time,
             updated_at=now_time,
+            input_version=input_version,
+            planning_input=effective_input,
+            input_fingerprint=effective_fingerprint,
         )
         db.add(pending)
         return pending
@@ -137,6 +220,25 @@ class TransferPending(Base):
             .limit(limit)
         ).scalars().all())
 
+    @classmethod
+    def list_by_states(cls, db: Session, *, states: tuple[str, ...],
+                       limit: Optional[int] = 5000) -> List["TransferPending"]:
+        """
+        按登记顺序列出多个可恢复持久状态的记录。
+        :param db: 数据库会话
+        :param states: 允许恢复的状态集合
+        :param limit: 单次读取上限
+        :return: 接纳记录列表
+        """
+        if not states:
+            return []
+        return list(db.execute(
+            select(cls)
+            .where(cls.state.in_(states))
+            .order_by(cls.created_at.asc(), cls.id.asc())
+            .limit(limit)
+        ).scalars().all())
+
     @classmethod
     def get_by_identity(cls, db: Session, *, storage: str,
                         src_path: str) -> Optional["TransferPending"]:
@@ -159,6 +261,90 @@ class TransferPending(Base):
             ).scalars().first(),
         )
 
+    @classmethod
+    def get_by_task_id(cls, db: Session, *, task_id: str) -> Optional["TransferPending"]:
+        """
+        按稳定任务标识查询一条持久登记。
+        :param db: 数据库会话
+        :param task_id: 稳定任务标识
+        :return: 接纳记录
+        """
+        if not task_id:
+            return None
+        return cast(
+            Optional["TransferPending"],
+            db.execute(select(cls).where(cls.task_id == task_id)).scalars().first(),
+        )
+
+    @classmethod
+    def checkpoint_plan(cls, db: Session, *, task_id: str,
+                        input_fingerprint: str, checkpoint_version: int,
+                        checkpoint_payload: dict[str, Any],
+                        source_states: tuple[str, ...], target_state: str,
+                        now_time: str) -> int:
+        """
+        以输入指纹为 CAS 条件原子保存计划并推进到已规划。
+        :param db: 数据库会话
+        :param task_id: 稳定任务标识
+        :param input_fingerprint: 规划输入规范 JSON 指纹
+        :param checkpoint_version: 检查点格式版本
+        :param checkpoint_payload: 完整有序计划 JSON
+        :param source_states: 允许推进检查点的起始状态
+        :param target_state: 检查点提交后的目标状态
+        :param now_time: 当前时间
+        :return: 更新的记录数
+        """
+        if (
+                not task_id
+                or not input_fingerprint
+                or not checkpoint_payload
+                or not source_states
+                or not target_state
+        ):
+            return 0
+        return execute_dml(
+            db,
+            update(cls)
+            .where(
+                cls.task_id == task_id,
+                cls.state.in_(source_states),
+                cls.input_fingerprint == input_fingerprint,
+            )
+            .values(
+                state=target_state,
+                checkpoint_version=checkpoint_version,
+                checkpoint_payload=checkpoint_payload,
+                planned_at=now_time,
+                last_error=None,
+                updated_at=now_time,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+
+    @classmethod
+    def record_planning_failure(cls, db: Session, *, task_id: str,
+                                error: str, now_time: str) -> int:
+        """
+        为接纳态或 provider 待执行任务记录规划失败，不改变其恢复状态。
+        :param db: 数据库会话
+        :param task_id: 稳定任务标识
+        :param error: 失败原因
+        :param now_time: 当前时间
+        :return: 更新的记录数
+        """
+        if not task_id:
+            return 0
+        return execute_dml(
+            db,
+            update(cls)
+            .where(
+                cls.task_id == task_id,
+                cls.state.in_(("accepted", "provider_pending")),
+            )
+            .values(last_error=error, updated_at=now_time),
+            execution_options={"synchronize_session": False},
+        )
+
     @classmethod
     def record_enqueue_failure(cls, db: Session, *, task_id: str,
                                error: str, now_time: str) -> int:

app/db/oper/transferpending.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import List, Optional, Tuple
+from typing import Any, List, Optional, Tuple
 
 from app.db.base import DbOper
 from app.db.models.transferpending import TransferPending
@@ -32,7 +32,9 @@ class TransferPendingOper(DbOper):
         )
 
     def stage_admit(self, *, task_id: str, storage: str, src_path: str,
-                    state: str, now_time: str) -> Optional[TransferPending]:
+                    state: str, now_time: str, input_version: int = 1,
+                    planning_input: Optional[dict[str, Any]] = None,
+                    input_fingerprint: Optional[str] = None) -> Optional[TransferPending]:
         """
         在当前会话中暂存一条持久接纳记录。
 
@@ -42,6 +44,9 @@ class TransferPendingOper(DbOper):
         :param src_path: 源文件路径
         :param state: 持久状态
         :param now_time: 当前时间
+        :param input_version: 规划输入格式版本
+        :param planning_input: 版本化规划输入 JSON
+        :param input_fingerprint: 规划输入规范 JSON 指纹
         :return: 接纳记录
         """
         return self._execute_sync_write(
@@ -52,6 +57,9 @@ class TransferPendingOper(DbOper):
                 src_path=src_path,
                 state=state,
                 now_time=now_time,
+                input_version=input_version,
+                planning_input=planning_input,
+                input_fingerprint=input_fingerprint,
             )
         )
 
@@ -71,6 +79,22 @@ class TransferPendingOper(DbOper):
             )
         ) or []
 
+    def list_by_states(self, *, states: tuple[str, ...],
+                       limit: Optional[int] = 5000) -> List[TransferPending]:
+        """
+        使用当前会话列出多个可恢复状态的记录。
+        :param states: 允许恢复的状态集合
+        :param limit: 单次读取上限
+        :return: ORM 接纳记录列表
+        """
+        return self._execute_sync_query(
+            lambda session: TransferPending.list_by_states(
+                session,
+                states=states,
+                limit=limit,
+            )
+        ) or []
+
     def get_by_identity(self, *, storage: str,
                         src_path: str) -> Optional[TransferPending]:
         """
@@ -87,6 +111,72 @@ class TransferPendingOper(DbOper):
             )
         )
 
+    def get_by_task_id(self, *, task_id: str) -> Optional[TransferPending]:
+        """
+        使用当前会话按稳定任务标识查询接纳记录。
+        :param task_id: 稳定任务标识
+        :return: 接纳记录
+        """
+        return self._execute_sync_query(
+            lambda session: TransferPending.get_by_task_id(
+                session,
+                task_id=task_id,
+            )
+        )
+
+    def stage_checkpoint_plan(
+            self,
+            *,
+            task_id: str,
+            input_fingerprint: str,
+            checkpoint_version: int,
+            checkpoint_payload: dict[str, Any],
+            source_states: tuple[str, ...],
+            target_state: str,
+            now_time: str,
+    ) -> int:
+        """
+        在当前会话中以输入指纹为条件暂存完整计划检查点。
+        :param task_id: 稳定任务标识
+        :param input_fingerprint: 规划输入规范 JSON 指纹
+        :param checkpoint_version: 检查点格式版本
+        :param checkpoint_payload: 完整有序计划 JSON
+        :param source_states: 允许执行 CAS 的起始状态
+        :param target_state: 检查点提交后的目标状态
+        :param now_time: 当前时间
+        :return: 更新的记录数
+        """
+        return self._execute_sync_write(
+            lambda session: TransferPending.checkpoint_plan(
+                session,
+                task_id=task_id,
+                input_fingerprint=input_fingerprint,
+                checkpoint_version=checkpoint_version,
+                checkpoint_payload=checkpoint_payload,
+                source_states=source_states,
+                target_state=target_state,
+                now_time=now_time,
+            )
+        )
+
+    def stage_record_planning_failure(self, *, task_id: str, error: str,
+                                      now_time: str) -> int:
+        """
+        在当前会话中记录规划失败并保持任务处于接纳态。
+        :param task_id: 稳定任务标识
+        :param error: 失败原因
+        :param now_time: 当前时间
+        :return: 更新的记录数
+        """
+        return self._execute_sync_write(
+            lambda session: TransferPending.record_planning_failure(
+                session,
+                task_id=task_id,
+                error=error,
+                now_time=now_time,
+            )
+        )
+
     def stage_record_enqueue_failure(self, *, task_id: str, error: str,
                                      now_time: str) -> int:
         """

app/modules/filemanager/module.py

@@ -1,28 +1,28 @@
 from pathlib import Path
-from typing import Any, Optional, List, Tuple, Union, Dict, Callable
-
-from app.runtime.settings import get_runtime_setting
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 
+from app.adapters.system.host import SystemUtils
+from app.application.directory import DirectoryHelper
+from app.application.messaging.message import MessageHelper
+from app.application.transfer import TransferPlanCheckpoint, TransferPlanningInput
 from app.domain.context import MediaInfo, MusicInfo
 from app.domain.meta.metabase import MetaBase
 from app.domain.meta.metamusic import MetaMusic
 from app.domain.metainfo import MetaInfo
-from app.application.directory import DirectoryHelper
-from app.application.messaging.message import MessageHelper
+from app.foundation import text as text_tools
 from app.foundation.reflection import ModuleHelper
-from app.runtime.log import logger
 from app.modules import _ModuleBase
 from app.modules.filemanager.storages import StorageBase
 from app.modules.filemanager.transhandler import TransHandler
-from app.schemas.transfer import TransferInfo
-from app.schemas.mediaserver import ExistMediaInfo
-from app.schemas.tmdb import TmdbEpisode
-from app.schemas.system import TransferDirectoryConf
-from app.schemas.workflow import FileItem
+from app.runtime.log import logger
+from app.runtime.settings import get_runtime_setting
 from app.schemas.file import StorageUsage
+from app.schemas.mediaserver import ExistMediaInfo
+from app.schemas.system import TransferDirectoryConf
+from app.schemas.tmdb import TmdbEpisode
+from app.schemas.transfer import TransferInfo
 from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType, ModuleType, OtherModulesType, StorageAction
-from app.adapters.system.host import SystemUtils
-from app.foundation import text as text_tools
+from app.schemas.workflow import FileItem
 
 
 class FileManagerModule(_ModuleBase):
@@ -345,7 +345,7 @@ class FileManagerModule(_ModuleBase):
 
     def get_file_item(self, storage: str, path: Path) -> Optional[FileItem]:
         """
-        根据路径获取文件项
+        根据路径严格获取文件项；普通兼容调度仍会隔离异常并投影为空结果。
         """
         if storage not in self._support_storages:
             return None
@@ -353,7 +353,7 @@ class FileManagerModule(_ModuleBase):
         if not storage_oper:
             logger.error(f"不支持 {storage} 的文件获取")
             return None
-        return storage_oper.get_item(path)
+        return storage_oper.get_item_strict(path)
 
     def get_parent_item(self, fileitem: FileItem) -> Optional[FileItem]:
         """
@@ -391,16 +391,24 @@ class FileManagerModule(_ModuleBase):
             previous_snapshot=previous_snapshot
         )
 
-    def transfer(self, fileitem: FileItem, meta: MetaBase, mediainfo: MediaInfo,
-                 target_directory: TransferDirectoryConf = None,
-                 target_storage: Optional[str] = None, target_path: Path = None,
-                 transfer_type: Optional[str] = None, scrape: Optional[bool] = None,
-                 library_type_folder: Optional[bool] = None, library_category_folder: Optional[bool] = None,
-                 episodes_info: List[TmdbEpisode] = None,
-                 source_oper: Callable = None, target_oper: Callable = None,
-                 preview: Optional[bool] = False) -> TransferInfo:
+    def plan_transfer(
+            self,
+            fileitem: FileItem,
+            meta: MetaBase,
+            mediainfo: Union[MediaInfo, MusicInfo],
+                      target_directory: Optional[TransferDirectoryConf] = None,
+                      target_storage: Optional[str] = None,
+                      target_path: Optional[Path] = None,
+                      transfer_type: Optional[str] = None, scrape: Optional[bool] = None,
+                      library_type_folder: Optional[bool] = None,
+                      library_category_folder: Optional[bool] = None,
+                      episodes_info: Optional[List[TmdbEpisode]] = None,
+                      source_oper: Optional[StorageBase] = None,
+                      preview: Optional[bool] = False,
+                      planning_input: Optional[TransferPlanningInput] = None,
+                      ) -> TransferPlanCheckpoint:
         """
-        文件整理
+        解析整理策略并生成零写副作用的冻结计划。
         :param fileitem:  文件信息
         :param meta: 预识别的元数据
         :param mediainfo:  识别的媒体信息
@@ -413,29 +421,26 @@ class FileManagerModule(_ModuleBase):
         :param library_category_folder: 是否按媒体类别创建目录
         :param episodes_info: 当前季的全部集信息
         :param source_oper: 源存储操作对象
-        :param target_oper: 目标存储操作对象
-        :return: {path, target_path, message}
+        :param planning_input: admission 阶段冻结的原始请求，传入时不得改写
+        :return: 可持久化的整理计划检查点
         """
         handler = TransHandler()
         # 检查目录路径
-        if fileitem.storage == "local" and not Path(fileitem.path).exists():
-            return TransferInfo(success=False,
-                                fileitem=fileitem,
-                                message=f"{fileitem.path} 不存在")
+        if (
+                fileitem.storage == "local"
+                and (not fileitem.path or not Path(fileitem.path).exists())
+        ):
+            raise ValueError(f"{fileitem.path} 不存在")
         # 目标路径不能是文件
         if target_path and target_path.is_file():
             logger.error(f"整理目标路径 {target_path} 是一个文件")
-            return TransferInfo(success=False,
-                                fileitem=fileitem,
-                                message=f"{target_path} 不是有效目录")
+            raise ValueError(f"{target_path} 不是有效目录")
         # 获取目标路径
         if target_directory:
             # 目标媒体库目录未设置
             if not target_directory.library_path:
                 logger.error(f"目标媒体库目录未设置，无法整理文件，源路径：{fileitem.path}")
-                return TransferInfo(success=False,
-                                    fileitem=fileitem,
-                                    message="目标媒体库目录未设置")
+                raise ValueError("目标媒体库目录未设置")
             # 整理方式
             if not transfer_type:
                 transfer_type = target_directory.transfer_type
@@ -467,56 +472,141 @@ class FileManagerModule(_ModuleBase):
             # 未找到有效的媒体库目录
             logger.error(
                 f"{mediainfo.type.value if mediainfo.type else '未知类型'} {mediainfo.title_year} 未找到有效的媒体库目录，无法整理文件，源路径：{fileitem.path}")
-            return TransferInfo(success=False,
-                                fileitem=fileitem,
-                                message="未找到有效的媒体库目录")
+            raise ValueError("未找到有效的媒体库目录")
         # 整理方式
         if not transfer_type:
-            logger.error(f"{target_directory.name} 未设置整理方式")
-            return TransferInfo(success=False,
-                                fileitem=fileitem,
-                                message=f"{target_directory.name} 未设置整理方式")
+            directory_name = target_directory.name if target_directory else "目标目录"
+            logger.error(f"{directory_name} 未设置整理方式")
+            raise ValueError(f"{directory_name} 未设置整理方式")
+        if target_path is None:
+            raise ValueError("整理规划缺少目标路径")
 
         # 源操作对象
+        source_storage = fileitem.storage or "local"
         if not source_oper:
-            source_oper = self.__get_storage_oper(fileitem.storage)
+            source_oper = self.__get_storage_oper(source_storage)
         if not source_oper:
-            return TransferInfo(success=False,
-                                message=f"不支持的存储类型：{fileitem.storage}",
-                                fileitem=fileitem,
-                                fail_list=[fileitem.path],
-                                transfer_type=transfer_type,
-                                need_notify=need_notify
-                                )
-        # 目的操作对象
-        if not target_oper:
-            if not target_storage:
-                target_storage = fileitem.storage
-            target_oper = self.__get_storage_oper(target_storage)
-        if not target_oper:
-            return TransferInfo(success=False,
-                                message=f"不支持的存储类型：{target_storage}",
-                                fileitem=fileitem,
-                                fail_list=[fileitem.path],
-                                transfer_type=transfer_type,
-                                need_notify=need_notify)
+            raise ValueError(f"不支持的存储类型：{source_storage}")
+
+        if not target_storage:
+            target_storage = source_storage
+        if planning_input is None:
+            planning_input = TransferPlanningInput(
+                source_fileitem=fileitem.model_dump(mode="json"),
+                meta=self._serialize_transfer_model(meta) if meta else None,
+                mediainfo=(
+                    self._serialize_transfer_model(mediainfo) if mediainfo else None
+                ),
+                target_directory=(
+                    target_directory.model_dump(mode="json")
+                    if target_directory
+                    else None
+                ),
+                target_storage=target_storage,
+                target_path=target_path.as_posix(),
+                requested_transfer_type=transfer_type,
+                media_type=mediainfo.type.value if mediainfo.type else None,
+                need_scrape=bool(need_scrape),
+                need_rename=bool(need_rename),
+                need_notify=bool(need_notify),
+                overwrite_mode=overwrite_mode,
+                episodes_info=tuple(
+                    episode.model_dump(mode="json") for episode in episodes_info or []
+                ),
+                preview=bool(preview),
+            )
 
-        # 整理
         logger.info(f"获取整理目标路径：【{target_storage}】{target_path}")
-        return handler.transfer_media(fileitem=fileitem,
-                                      in_meta=meta,
-                                      mediainfo=mediainfo,
-                                      target_storage=target_storage,
-                                      target_path=target_path,
-                                      transfer_type=transfer_type,
-                                      need_scrape=need_scrape,
-                                      need_rename=need_rename,
-                                      need_notify=need_notify,
-                                      overwrite_mode=overwrite_mode,
-                                      episodes_info=episodes_info,
-                                      preview=preview,
-                                      source_oper=source_oper,
-                                      target_oper=target_oper)
+        return handler.plan_transfer(
+            planning_input,
+            meta=meta,
+            mediainfo=mediainfo,
+            source_oper=source_oper,
+            target_storage=target_storage,
+            target_path=target_path,
+            transfer_type=transfer_type,
+            need_scrape=bool(need_scrape),
+            need_rename=bool(need_rename),
+            need_notify=bool(need_notify),
+            overwrite_mode=overwrite_mode,
+            episodes_info=episodes_info,
+            preview=bool(preview),
+        )
+
+    def execute_transfer_plan(
+            self,
+            checkpoint: TransferPlanCheckpoint,
+            *,
+            meta: MetaBase,
+            mediainfo: Union[MediaInfo, MusicInfo],
+            source_oper: Optional[StorageBase] = None,
+            target_oper: Optional[StorageBase] = None,
+            cleanup_media_file: Optional[Callable[[FileItem], bool]] = None,
+    ) -> TransferInfo:
+        """解析存储适配器并通过统一删除能力执行已冻结计划。"""
+        source_fileitem = FileItem(**checkpoint.planning_input.source_fileitem)
+        cleanup_before_transfer = None
+        cleanup_payload = checkpoint.planning_input.options.get(
+            "cleanup_dest_fileitem"
+        )
+        if (
+                isinstance(cleanup_payload, dict)
+                and not checkpoint.preview
+                and not checkpoint.pre_execution_cleanup_completed
+        ):
+            cleanup_fileitem = FileItem.model_validate(cleanup_payload)
+            if not cleanup_media_file:
+                raise RuntimeError("整理计划缺少统一媒体删除兼容能力")
+
+            def cleanup_before_transfer() -> None:
+                """拦截通过后委托统一能力治理旧目标及父空目录。"""
+                if not cleanup_media_file(cleanup_fileitem):
+                    raise RuntimeError(
+                        f"{cleanup_fileitem.path} 删除失败，整理计划保留待重试"
+                    )
+
+        source_storage = source_fileitem.storage or "local"
+        if not source_oper:
+            source_oper = self.__get_storage_oper(source_storage)
+        if not source_oper:
+            return TransferInfo(
+                success=False,
+                message=f"不支持的存储类型：{source_storage}",
+                fileitem=source_fileitem,
+                fail_list=[source_fileitem.path],
+                transfer_type=checkpoint.resolved_transfer_type,
+                need_notify=checkpoint.need_notify,
+            )
+        if not target_oper:
+            target_oper = self.__get_storage_oper(checkpoint.target_storage)
+        if not target_oper:
+            return TransferInfo(
+                success=False,
+                message=f"不支持的存储类型：{checkpoint.target_storage}",
+                fileitem=source_fileitem,
+                fail_list=[source_fileitem.path],
+                transfer_type=checkpoint.resolved_transfer_type,
+                need_notify=checkpoint.need_notify,
+            )
+        return TransHandler().execute_transfer_plan(
+            checkpoint,
+            meta=meta,
+            mediainfo=mediainfo,
+            source_oper=source_oper,
+            target_oper=target_oper,
+            cleanup_before_transfer=cleanup_before_transfer,
+        )
+
+    @staticmethod
+    def _serialize_transfer_model(value: object) -> dict[str, Any]:
+        """校验旧领域对象的动态序列化结果，避免 Any 穿透规划边界。"""
+        serializer = getattr(value, "to_dict", None)
+        if not callable(serializer):
+            raise TypeError(f"{type(value).__name__} 不支持整理快照序列化")
+        payload = serializer()
+        if not isinstance(payload, dict):
+            raise TypeError(f"{type(value).__name__} 返回了无效整理快照")
+        return payload
 
     @staticmethod
     def _build_library_lookup_meta(

app/runtime/extensions/module/contracts.py

@@ -124,6 +124,29 @@ _METHOD_CONTRACTS = {
     "rename_file": ModuleMethodContract(family="storage", input_contract="StorageRenameRequest", result_contract="bool | FileItem", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("fileitem", "name")),
     "storage_manage": ModuleMethodContract(family="storage", input_contract="StorageManageRequest", result_contract="StorageProviderResult", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("storage", "action")),
     "snapshot_storage": ModuleMethodContract(family="storage", input_contract="StorageSnapshotRequest", result_contract="dict[str, dict] | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("storage", "path", "last_snapshot_time", "max_depth", "previous_snapshot")),
+    "plan_transfer": ModuleMethodContract(
+        family="storage",
+        input_contract="TransferPlanningInput",
+        result_contract="TransferPlanCheckpoint | None",
+        aggregation=ModuleResultAggregation.FIRST_NON_EMPTY,
+        required_parameters=(
+            "fileitem", "meta", "mediainfo", "target_directory", "target_storage",
+            "target_path", "transfer_type", "scrape", "library_type_folder",
+            "library_category_folder", "episodes_info", "source_oper", "preview",
+            "planning_input",
+        ),
+        public_to_plugins=False,
+    ),
+    "execute_transfer_plan": ModuleMethodContract(
+        family="storage",
+        input_contract="TransferPlanCheckpoint",
+        result_contract="TransferInfo | None",
+        aggregation=ModuleResultAggregation.FIRST_NON_EMPTY,
+        required_parameters=(
+            "checkpoint", "meta", "mediainfo", "source_oper", "target_oper",
+        ),
+        public_to_plugins=False,
+    ),
     "transfer": ModuleMethodContract(family="storage", input_contract="TransferRequest", result_contract="TransferInfo | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("fileitem", "meta", "mediainfo", "target_directory", "target_storage", "target_path", "transfer_type", "scrape", "library_type_folder", "library_category_folder", "episodes_info", "source_oper", "target_oper", "preview")),
     "load_category_config": ModuleMethodContract(family="category", input_contract="CategoryConfigReadRequest", result_contract="CategoryConfig | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY),
     "clear_cache": ModuleMethodContract(family="category", input_contract="CacheClearRequest", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, plugin_short_circuit=False),

app/runtime/extensions/module/dispatcher.py

@@ -4,13 +4,12 @@ from __future__ import annotations
 
 import inspect
 from collections.abc import Callable, Mapping
+from dataclasses import dataclass
 from enum import StrEnum
 from typing import Any, Protocol, cast
 
 from app.foundation.reflection import ObjectUtils
 from app.runtime.execution import run_in_threadpool_to_completion
-from app.runtime.log import logger
-from app.runtime.observability import observe_duration, record_metric
 from app.runtime.extensions.module.contracts import (
     ModuleResultAggregation,
     diagnose_module_callable,
@@ -18,6 +17,8 @@ from app.runtime.extensions.module.contracts import (
     get_module_method_contract,
     is_explicit_module_method,
 )
+from app.runtime.log import logger
+from app.runtime.observability import observe_duration, record_metric
 from app.schemas.exception import RateLimitExceededException
 
 
@@ -41,6 +42,50 @@ ModuleErrorHandler = Callable[..., None]
 AsyncFunctionRunner = Callable[..., Any]
 
 
+class FrozenModuleProviderMissingError(LookupError):
+    """表示冻结插件 provider 已无法在当前精确目录中解析。"""
+
+
+@dataclass(frozen=True, slots=True)
+class FrozenPluginProviderRef:
+    """保存可持久化的插件 provider 身份及其冻结方法。"""
+
+    plugin_id: str
+    plugin_name: str
+    method: str
+
+    def __post_init__(self) -> None:
+        """拒绝无法精确解析的空 provider 身份。"""
+        if not self.plugin_id or not self.plugin_name or not self.method:
+            raise ValueError("冻结插件 provider 缺少 id、名称或方法")
+
+    def to_payload(self) -> dict[str, str]:
+        """生成可直接写入 JSON 的稳定 provider 引用。"""
+        return {
+            "plugin_id": self.plugin_id,
+            "plugin_name": self.plugin_name,
+            "method": self.method,
+        }
+
+    @classmethod
+    def from_payload(cls, payload: Mapping[str, Any]) -> "FrozenPluginProviderRef":
+        """从持久化映射恢复并校验 provider 引用。"""
+        return cls(
+            plugin_id=str(payload.get("plugin_id") or ""),
+            plugin_name=str(payload.get("plugin_name") or ""),
+            method=str(payload.get("method") or ""),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class _PluginProvider:
+    """绑定一次目录解析得到的插件身份和可调用对象。"""
+
+    plugin_id: str
+    plugin_name: str
+    func: Callable[..., Any]
+
+
 class _ProviderCallMode(StrEnum):
     """描述当前 provider 应采用的兼容调用方式。"""
 
@@ -77,14 +122,72 @@ class ModuleInvocationDispatcher:
             return all(value is None for value in result)
         return result is None
 
+    def freeze_plugin_providers(
+        self,
+        method: str,
+    ) -> tuple[FrozenPluginProviderRef, ...]:
+        """按当前插件目录顺序冻结实现指定方法的精确 provider 引用。"""
+        return tuple(
+            FrozenPluginProviderRef(
+                plugin_id=provider.plugin_id,
+                plugin_name=provider.plugin_name,
+                method=method,
+            )
+            for provider in self._collect_plugin_providers(method, error_kwargs={})
+        )
+
+    def execute_frozen_plugin_providers(
+        self,
+        method: str,
+        providers: tuple[FrozenPluginProviderRef, ...],
+        *args: Any,
+        initial_result: Any = None,
+        before_invoke: Callable[[], None] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        """解析全部冻结引用后运行前置钩子，并严格执行原顺序 provider。"""
+        resolved = self._resolve_frozen_plugin_providers(method, providers)
+        if resolved and before_invoke is not None:
+            before_invoke()
+        return self._execute_plugin_provider_sequence(
+            method,
+            initial_result,
+            resolved,
+            *args,
+            strict_errors=True,
+            **kwargs,
+        )
+
     def dispatch(self, method: str, *args: Any, **kwargs: Any) -> Any:
         """先执行插件模块，再按优先级执行宿主模块。"""
+        return self._dispatch(method, *args, strict_errors=False, **kwargs)
+
+    def dispatch_strict(self, method: str, *args: Any, **kwargs: Any) -> Any:
+        """执行模块并传播 provider 异常，使空结果与查询失败保持可区分。"""
+        return self._dispatch(method, *args, strict_errors=True, **kwargs)
+
+    def _dispatch(
+        self,
+        method: str,
+        *args: Any,
+        strict_errors: bool,
+        **kwargs: Any,
+    ) -> Any:
+        """按统一聚合规则调度，并由调用方选择是否隔离 provider 异常。"""
         contract = get_module_method_contract(method)
         logger.debug("模块方法契约：%s -> %s", method, contract.family)
-        with observe_duration(
-            "module.provider.duration", method=method, provider_type="plugin"
-        ):
-            result = self.execute_plugin_modules(method, None, *args, **kwargs)
+        result = None
+        if contract.public_to_plugins:
+            with observe_duration(
+                "module.provider.duration", method=method, provider_type="plugin"
+            ):
+                result = self.execute_plugin_modules(
+                    method,
+                    None,
+                    *args,
+                    strict_errors=strict_errors,
+                    **kwargs,
+                )
         if (
             contract.plugin_short_circuit
             and not self.is_valid_empty(result)
@@ -94,21 +197,29 @@ class ModuleInvocationDispatcher:
         with observe_duration(
             "module.provider.duration", method=method, provider_type="system"
         ):
-            return self.execute_system_modules(method, result, *args, **kwargs)
+            return self.execute_system_modules(
+                method,
+                result,
+                *args,
+                strict_errors=strict_errors,
+                **kwargs,
+            )
 
     async def async_dispatch(self, method: str, *args: Any, **kwargs: Any) -> Any:
         """以与同步路径相同的聚合规则执行同步或异步模块方法。"""
         contract = get_module_method_contract(method)
         logger.debug("异步模块方法契约：%s -> %s", method, contract.family)
-        with observe_duration(
-            "module.provider.duration", method=method, provider_type="plugin"
-        ):
-            result = await self.async_execute_plugin_modules(
-                method,
-                None,
-                *args,
-                **kwargs,
-            )
+        result = None
+        if contract.public_to_plugins:
+            with observe_duration(
+                "module.provider.duration", method=method, provider_type="plugin"
+            ):
+                result = await self.async_execute_plugin_modules(
+                    method,
+                    None,
+                    *args,
+                    **kwargs,
+                )
         if (
             contract.plugin_short_circuit
             and not self.is_valid_empty(result)
@@ -130,37 +241,130 @@ class ModuleInvocationDispatcher:
         method: str,
         result: Any,
         *args: Any,
+        strict_errors: bool = False,
         **kwargs: Any,
     ) -> Any:
         """同步执行插件方法，保留插件顺序、短路和列表合并语义。"""
-        aggregation = get_module_method_contract(method).aggregation
+        providers = self._collect_plugin_providers(
+            method,
+            error_kwargs=kwargs,
+            strict_errors=strict_errors,
+        )
+        return self._execute_plugin_provider_sequence(
+            method,
+            result,
+            providers,
+            *args,
+            strict_errors=strict_errors,
+            **kwargs,
+        )
+
+    def _collect_plugin_providers(
+        self,
+        method: str,
+        *,
+        error_kwargs: Mapping[str, Any],
+        strict_errors: bool = False,
+    ) -> tuple[_PluginProvider, ...]:
+        """从同一插件目录快照收集 provider，并隔离损坏的方法表。"""
+        providers = []
         for plugin, module_dict in self._plugin_catalog.get_plugin_modules().items():
             plugin_id, plugin_name = plugin
             try:
-                # 防御坏插件把方法表声明成非映射类型，避免击穿整个模块调度
                 if not isinstance(module_dict, Mapping):
                     raise TypeError(
-                        f"插件 {plugin_id} 的模块声明必须是映射，实际是 {type(module_dict).__name__}"
+                        f"插件 {plugin_id} 的模块声明必须是映射，实际是 "
+                        f"{type(module_dict).__name__}"
                     )
                 func = module_dict.get(method)
                 if not func:
                     continue
+                providers.append(
+                    _PluginProvider(
+                        plugin_id=plugin_id,
+                        plugin_name=plugin_name,
+                        func=func,
+                    )
+                )
+            except Exception as err:
+                self._record_timeout(method, "plugin", err)
+                self._plugin_error_handler(
+                    err,
+                    plugin_id,
+                    plugin_name,
+                    method,
+                    **error_kwargs,
+                )
+                if strict_errors:
+                    raise
+        return tuple(providers)
+
+    def _resolve_frozen_plugin_providers(
+        self,
+        method: str,
+        providers: tuple[FrozenPluginProviderRef, ...],
+    ) -> tuple[_PluginProvider, ...]:
+        """一次性精确解析全部冻结引用，缺失时在执行任何副作用前失败。"""
+        module_catalog = self._plugin_catalog.get_plugin_modules()
+        resolved = []
+        for provider in providers:
+            if provider.method != method:
+                raise FrozenModuleProviderMissingError(
+                    f"冻结插件 provider 方法不匹配："
+                    f"{provider.plugin_id}/{provider.plugin_name} "
+                    f"冻结为 {provider.method}，请求执行 {method}"
+                )
+            module_dict = module_catalog.get(
+                (provider.plugin_id, provider.plugin_name)
+            )
+            func = module_dict.get(method) if isinstance(module_dict, Mapping) else None
+            if not callable(func):
+                raise FrozenModuleProviderMissingError(
+                    f"冻结插件 provider 已缺失："
+                    f"{provider.plugin_id}/{provider.plugin_name}.{method}"
+                )
+            resolved.append(
+                _PluginProvider(
+                    plugin_id=provider.plugin_id,
+                    plugin_name=provider.plugin_name,
+                    func=func,
+                )
+            )
+        return tuple(resolved)
+
+    def _execute_plugin_provider_sequence(
+        self,
+        method: str,
+        result: Any,
+        providers: tuple[_PluginProvider, ...],
+        *args: Any,
+        strict_errors: bool = False,
+        **kwargs: Any,
+    ) -> Any:
+        """按统一契约执行已解析插件序列，并按调用模式处理 provider 故障。"""
+        aggregation = get_module_method_contract(method).aggregation
+        for provider in providers:
+            try:
                 self._record_legacy_hit(
                     method,
                     caller_type="plugin",
                     abi_source="third_party_plugin",
                 )
-                self._diagnose_callable(method, func, f"插件 {plugin_id}")
-                logger.info("请求插件 %s 执行：%s ...", plugin_name, method)
+                self._diagnose_callable(
+                    method,
+                    provider.func,
+                    f"插件 {provider.plugin_id}",
+                )
+                logger.info("请求插件 %s 执行：%s ...", provider.plugin_name, method)
                 call_mode = self._provider_call_mode(
                     aggregation,
                     result,
-                    func,
+                    provider.func,
                     allow_relay=False,
                 )
                 if call_mode is _ProviderCallMode.STOP:
                     break
-                provider_result = func(*args, **kwargs)
+                provider_result = provider.func(*args, **kwargs)
                 self._diagnose_result(method, provider_result, "plugin")
                 result = self._aggregate_provider_result(
                     result,
@@ -172,19 +376,23 @@ class ModuleInvocationDispatcher:
                 self._rate_limit_handler(
                     err,
                     "插件",
-                    plugin_id,
+                    provider.plugin_id,
                     method,
                     **kwargs,
                 )
+                if strict_errors:
+                    raise
             except Exception as err:
                 self._record_timeout(method, "plugin", err)
                 self._plugin_error_handler(
                     err,
-                    plugin_id,
-                    plugin_name,
+                    provider.plugin_id,
+                    provider.plugin_name,
                     method,
                     **kwargs,
                 )
+                if strict_errors:
+                    raise
         return result
 
     async def async_execute_plugin_modules(
@@ -254,6 +462,7 @@ class ModuleInvocationDispatcher:
         method: str,
         result: Any,
         *args: Any,
+        strict_errors: bool = False,
         **kwargs: Any,
     ) -> Any:
         """同步执行按优先级排序的宿主模块，并支持签名接力。"""
@@ -301,6 +510,8 @@ class ModuleInvocationDispatcher:
                     method,
                     **kwargs,
                 )
+                if strict_errors:
+                    raise
             except Exception as err:
                 self._record_timeout(method, "system", err)
                 self._system_error_handler(
@@ -310,6 +521,8 @@ class ModuleInvocationDispatcher:
                     method,
                     **kwargs,
                 )
+                if strict_errors:
+                    raise
         return result
 
     async def async_execute_system_modules(

app/startup/initializers/modules.py

@@ -1,7 +1,7 @@
 import asyncio
 import inspect
 import sys
-from typing import Callable
+from typing import Any, Callable
 
 from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper
 from app.application.plugin.transaction import (
@@ -241,6 +241,13 @@ async def _async_get_workflow(workflow_id: int):
     return await WorkflowOper().async_get(workflow_id)
 
 
+def _execute_legacy_transfer_command(**kwargs: Any) -> Any:
+    """把旧 Chain ABI 延迟转入唯一 TransferChain durable command。"""
+    from app.chain.transfer import TransferChain
+
+    return TransferChain().execute_legacy_transfer_command(**kwargs)
+
+
 def _build_chain_runtime_context() -> ChainRuntimeContext:
     """在启动组合根创建 Chain 所需的运行时对象和数据端口。"""
     return ChainRuntimeContext(
@@ -255,6 +262,7 @@ def _build_chain_runtime_context() -> ChainRuntimeContext:
             send_callback=callback
         ),
         module_dispatcher_factory=ModuleInvocationDispatcher,
+        legacy_transfer_command=_execute_legacy_transfer_command,
         configuration=build_chain_runtime_config(legacy_settings),
         data_ports=get_chain_data_ports(),
         durable_event_writer=TransactionalChainDurableEventWriter(SessionFactory),