fix(transfer): resolve durable failure recovery issues

2026-11-04 06:15:14 +08:00 · 2026-09-01 07:11:06 +08:00
parent 6534eca295
commit 94716484b7
22 changed files with 762 additions and 80 deletions
--- a/app/api/dependencies/history.py
+++ b/app/api/dependencies/history.py
@@ -91,6 +91,7 @@ def get_transfer_history_mutation_command(
    return TransferHistoryMutationCommand(
        repository=runtime.history.transfer_mutation_repository(db),
        download_repository=runtime.history.download_repository(db),
+        transfer_execution_repository=runtime.history.transfer_execution_repository,
        unit_of_work=runtime.persistence.sync_transaction(db),
        file_item_factory=lambda payload: _SchemaFileItem(**payload),
        file_exists=storage_chain.exists,
--- a/app/application/historymutation.py
+++ b/app/application/historymutation.py
@@ -6,6 +6,10 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, Optional, Protocol

+from app.application.transfer.execution import (
+    TransferExecutionCommand,
+    TransferExecutionRepository,
+)
 from app.schemas.common import JsonData
 from app.schemas.history import TransferHistoryDeleteResult, TransferHistoryDeleteStep

@@ -34,6 +38,11 @@ class TransferHistoryMutationRecord(Protocol):
        """返回持久整理任务标识。"""
        ...

+    @property
+    def transfer_settlement_revision(self) -> Optional[int]:
+        """返回持久整理失败回执的结算版本。"""
+        ...
+
    @property
    def src(self) -> Optional[str]:
        """返回源路径。"""
@@ -132,6 +141,7 @@ class TransferHistoryMutationCommand:
        *,
        repository: TransferHistoryMutationRepository,
        download_repository: DownloadFileMutationRepository,
+        transfer_execution_repository: TransferExecutionRepository,
        unit_of_work: HistoryUnitOfWork,
        file_item_factory: Callable[[dict[str, JsonData]], Any],
        delete_media_file: Callable[[Any], bool],
@@ -142,6 +152,7 @@ class TransferHistoryMutationCommand:
        """保存历史事务、存储删除、事件和失败状态清理端口。"""
        self._repository = repository
        self._download_repository = download_repository
+        self._transfer_execution_repository = transfer_execution_repository
        self._unit_of_work = unit_of_work
        self._file_item_factory = file_item_factory
        self._file_exists = file_exists or (lambda _fileitem: True)
@@ -166,12 +177,28 @@ class TransferHistoryMutationCommand:
                message="记录不存在",
            )
        if history.transfer_task_id:
-            return TransferHistoryDeleteResult(
-                source=TransferHistoryDeleteStep(status="not_requested"),
-                destination=TransferHistoryDeleteStep(status="not_requested"),
-                history="retained",
-                message="持久整理失败记录不可删除，请使用重试或人工复核入口",
+            settlement_revision = history.transfer_settlement_revision
+            if not settlement_revision:
+                return TransferHistoryDeleteResult(
+                    source=TransferHistoryDeleteStep(status="not_requested"),
+                    destination=TransferHistoryDeleteStep(status="not_requested"),
+                    history="retained",
+                    message="持久整理失败记录缺少结算版本，请刷新后重试",
+                )
+            discard = TransferExecutionCommand(
+                self._transfer_execution_repository
+            ).discard_failed(
+                task_id=history.transfer_task_id,
+                history_id=history_id,
+                settlement_revision=settlement_revision,
            )
+            if not discard.discarded:
+                return TransferHistoryDeleteResult(
+                    source=TransferHistoryDeleteStep(status="not_requested"),
+                    destination=TransferHistoryDeleteStep(status="not_requested"),
+                    history="retained",
+                    message=discard.message,
+                )

        destination_result = self._delete_file(
            history.dest_fileitem if delete_destination else None,
--- a/app/application/transfer/execution.py
+++ b/app/application/transfer/execution.py
@@ -435,6 +435,15 @@ class TransferRetryRequestResult:
    message: str


+@dataclass(frozen=True, slots=True)
+class TransferFailureDiscardResult:
+    """描述终态失败任务是否已解除历史绑定并删除执行证据。"""
+
+    discarded: bool
+    state: Optional[TransferExecutionState]
+    message: str
+
+
@dataclass(frozen=True, slots=True)
 class TransferManualReviewResult:
    """描述一次已持久审计的人工判定及后续调度状态。"""
@@ -595,6 +604,15 @@ class TransferExecutionRepository(Protocol):
    ) -> TransferRetryRequestResult:
        """仅把 FAILED 终态转入到期可 claim 的 retry_wait。"""

+    def discard_failed(
+            self,
+            *,
+            task_id: str,
+            history_id: int,
+            settlement_revision: int,
+    ) -> TransferFailureDiscardResult:
+        """仅放弃与指定失败回执完全匹配的无租约 FAILED 任务。"""
+
    def resolve_manual_review(
            self,
            *,
@@ -810,6 +828,22 @@ class TransferExecutionCommand:
            requested_by=requested_by,
        )

+    def discard_failed(
+            self,
+            *,
+            task_id: str,
+            history_id: int,
+            settlement_revision: int,
+    ) -> TransferFailureDiscardResult:
+        """放弃确定失败任务，使对应历史恢复为普通可维护记录。"""
+        if not task_id or history_id <= 0 or settlement_revision <= 0:
+            raise ValueError("放弃失败整理任务缺少任务、历史或结算版本")
+        return self._repository.discard_failed(
+            task_id=task_id,
+            history_id=history_id,
+            settlement_revision=settlement_revision,
+        )
+
    def resolve_manual_review(
            self,
            *,
@@ -924,6 +958,7 @@ __all__ = [
    "TransferExecutionCommand",
    "TransferExecutionConflictError",
    "TransferExecutionError",
+    "TransferFailureDiscardResult",
    "TransferExecutionLeaseLostError",
    "TransferOperationObservation",
    "TransferOperationObservationState",
--- a/app/application/transfer/workflow.py
+++ b/app/application/transfer/workflow.py
@@ -705,6 +705,10 @@ class TransferAdmissionConflictError(ValueError):
    """同一源文件以不同规划输入重复准入时抛出的冲突错误。"""


+class TransferPlanningRejectedError(ValueError):
+    """规划输入缺少业务必需信息且继续自动重试不会改变结果。"""
+
+
 class TransferPlanningStateError(RuntimeError):
    """计划检查点无法从当前持久状态推进时抛出的状态错误。"""

--- a/app/chain/base.py
+++ b/app/chain/base.py
@@ -951,7 +951,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
        """调用文件管理模块生成无文件写入的冻结整理计划。"""
        return cast(
            "Optional[TransferPlanCheckpoint]",
-            self.run_module(
+            self.run_module_strict(
                "plan_transfer",
                fileitem=fileitem,
                meta=meta,
--- a/app/chain/transfer/plan.py
+++ b/app/chain/transfer/plan.py
@@ -17,6 +17,7 @@ from app.application.transfer.workflow import (
    TransferLeaseLostError,
    TransferPlanCheckpoint,
    TransferPlanningInput,
+    TransferPlanningRejectedError,
    TransferPlanningStateError,
    TransferProviderInvocationSnapshot,
    TransferProviderReference,
@@ -274,13 +275,34 @@ class TransferPlanningOwner(_TransferOwnerBase):
            )
        planning_input = task.planning_input or self._TransferChain__build_planning_input(task)
        task.bind_planning_input(planning_input)
+        checkpoint = self._TransferChain__build_planning_rejection_checkpoint(
+            task,
+            error=error,
+            planning_input=planning_input,
+        )
+        persisted = self._TransferChain__persist_transfer_checkpoint(
+            task,
+            planning_input=planning_input,
+            checkpoint=checkpoint,
+        )
+        task.bind_plan_checkpoint(persisted)
+        return self._plan_checkpoint_and_execute(task)
+
+    def _TransferChain__build_planning_rejection_checkpoint(
+            self,
+            task: TransferTask,
+            *,
+            error: str,
+            planning_input: TransferPlanningInput,
+    ) -> TransferPlanCheckpoint:
+        """把确定性的宿主规划错误冻结为可重放的零步骤失败计划。"""
        source_path = task.fileitem.path
        resolved_transfer_type = (
            task.transfer_type
            or planning_input.requested_transfer_type
            or "copy"
        )
-        checkpoint = TransferPlanCheckpoint(
+        return TransferPlanCheckpoint(
            planning_input=planning_input,
            target_storage=task.fileitem.storage,
            root_target_path=source_path,
@@ -300,13 +322,6 @@ class TransferPlanningOwner(_TransferOwnerBase):
            overwrite_mode=planning_input.overwrite_mode,
            rejection_error=error,
        )
-        persisted = self._TransferChain__persist_transfer_checkpoint(
-            task,
-            planning_input=planning_input,
-            checkpoint=checkpoint,
-        )
-        task.bind_plan_checkpoint(persisted)
-        return self._plan_checkpoint_and_execute(task)

    def _TransferChain__execute_planning_rejection(
            self,
@@ -675,25 +690,32 @@ class TransferPlanningOwner(_TransferOwnerBase):
            source_oper: Any,
    ) -> TransferPlanCheckpoint:
        """在 provider 未接管时生成纯宿主计划，不执行任何文件写入。"""
-        checkpoint = cast(
-            Optional[TransferPlanCheckpoint],
-            self.plan_transfer(
-                fileitem=task.fileitem,
-                meta=cast(MetaBase, task.meta),
-                mediainfo=cast(Union[MediaInfo, MusicInfo], task.mediainfo),
-                target_directory=task.target_directory,
-                target_storage=task.target_storage,
-                target_path=task.target_path,
-                transfer_type=task.transfer_type,
-                episodes_info=task.episodes_info,
-                scrape=task.scrape,
-                library_type_folder=task.library_type_folder,
-                library_category_folder=task.library_category_folder,
-                source_oper=source_oper,
-                preview=bool(task.preview),
+        try:
+            checkpoint = cast(
+                Optional[TransferPlanCheckpoint],
+                self.plan_transfer(
+                    fileitem=task.fileitem,
+                    meta=cast(MetaBase, task.meta),
+                    mediainfo=cast(Union[MediaInfo, MusicInfo], task.mediainfo),
+                    target_directory=task.target_directory,
+                    target_storage=task.target_storage,
+                    target_path=task.target_path,
+                    transfer_type=task.transfer_type,
+                    episodes_info=task.episodes_info,
+                    scrape=task.scrape,
+                    library_type_folder=task.library_type_folder,
+                    library_category_folder=task.library_category_folder,
+                    source_oper=source_oper,
+                    preview=bool(task.preview),
+                    planning_input=planning_input,
+                ),
+            )
+        except TransferPlanningRejectedError as error:
+            return self._TransferChain__build_planning_rejection_checkpoint(
+                task,
+                error=str(error),
                planning_input=planning_input,
-            ),
-        )
+            )
        if checkpoint is None:
            raise RuntimeError("文件整理模块未返回规划检查点")
        return checkpoint
--- a/app/chain/transfer/records.py
+++ b/app/chain/transfer/records.py
@@ -13,6 +13,7 @@ from app.application.history import (
    resolve_history,
 )
 from app.application.transfer.execution import (
+    TransferExecutionCommand,
    TransferExecutionRepository,
 )
 from app.chain._contracts import TransferMixinHost
@@ -477,8 +478,8 @@ class ManualHistoryMixin(_TransferOwnerBase):
    ) -> Optional[Tuple[bool, str]]:
        """将 durable 历史重试交还持久调度器，旧历史返回 ``None``。

-        durable 任务的历史、目标文件和步骤证据共同描述一次可恢复执行。任何历史
-        入口都只能登记重试意图，不能删除这些证据后重新准入一条并行任务。
+        普通重试和 AI 接管只登记重试意图；显式重新整理由
+        ``_delete_manual_transfer_history`` 先放弃确定失败任务，再重新准入。

        :param history: 整理历史
        :param requested_by: 发起重试的稳定入口身份
@@ -495,15 +496,25 @@ class ManualHistoryMixin(_TransferOwnerBase):
            history: TransferHistorySnapshot,
            transfer_history_oper: TransferHistoryRepository,
    ) -> Tuple[bool, str]:
-        """删除手动重整历史；非成功移动记录同时清理可能存在的旧目标。"""
-        durable_retry = self._request_durable_transfer_retry(
-            history,
-            requested_by="manual_reorganize",
-        )
-        if durable_retry is not None:
-            _, message = durable_retry
-            # 返回 False 阻止旧 caller 把“已登记”误当成“历史已删除”并继续执行。
-            return False, message
+        """删除手动重整历史；失败 durable 回执先原子放弃再清理旧目标。"""
+        task_id = getattr(history, "transfer_task_id", None)
+        if task_id:
+            settlement_revision = getattr(
+                history,
+                "transfer_settlement_revision",
+                None,
+            )
+            if not settlement_revision:
+                return False, "持久整理失败记录缺少结算版本，请刷新后重试"
+            discard = TransferExecutionCommand(
+                self.transfer_execution_repository
+            ).discard_failed(
+                task_id=task_id,
+                history_id=history.id,
+                settlement_revision=settlement_revision,
+            )
+            if not discard.discarded:
+                return False, discard.message
        if (
                history.dest_fileitem
                and not ManualHistoryMixin._is_successful_move_history(history)
--- a/app/chain/transfer/workflow.py
+++ b/app/chain/transfer/workflow.py
@@ -710,20 +710,21 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                                reorganize or not transferd.status
                        )
                        if should_reorganize:
-                            durable_retry = self._request_durable_transfer_retry(
-                                transferd,
-                                requested_by="manual_reorganize",
-                            )
-                            if durable_retry is not None:
-                                accepted, message = durable_retry
-                                if accepted:
-                                    logger.info(message)
-                                else:
-                                    all_success = False
-                                    logger.error(message)
-                                    err_msgs.append(message)
-                                # durable 历史只登记唯一调度重试，不在当前调用重新准入。
-                                continue
+                            if not reorganize:
+                                durable_retry = self._request_durable_transfer_retry(
+                                    transferd,
+                                    requested_by="manual_reorganize",
+                                )
+                                if durable_retry is not None:
+                                    accepted, message = durable_retry
+                                    if accepted:
+                                        logger.info(message)
+                                    else:
+                                        all_success = False
+                                        logger.error(message)
+                                        err_msgs.append(message)
+                                    # 普通失败重试仍由唯一 durable 调度器继续原计划。
+                                    continue
                            state, message = self._delete_manual_transfer_history(
                                history=transferd,
                                transfer_history_oper=transfer_history_oper,
--- a/app/db/adapters/transfer/execution.py
+++ b/app/db/adapters/transfer/execution.py
@@ -18,6 +18,7 @@ from app.application.transfer.execution import (
    TransferExecutionSnapshot,
    TransferExecutionState,
    TransferExecutionStep,
+    TransferFailureDiscardResult,
    TransferManualReviewDecision,
    TransferManualReviewPage,
    TransferManualReviewResult,
@@ -42,6 +43,7 @@ from app.db.models.transferexecutionstep import (
 )
 from app.db.models.transferpending import TransferPending
 from app.db.oper.transferexecutionstep import TransferExecutionStepOper
+from app.db.oper.transferhistory import TransferHistoryOper
 from app.db.oper.transferpending import TransferPendingOper
 from app.db.uow import SqlAlchemyUnitOfWork

@@ -1197,6 +1199,110 @@ class TransactionalTransferExecutionRepository:
                self._rollback(transaction)
                raise

+    def discard_failed(
+            self,
+            *,
+            task_id: str,
+            history_id: int,
+            settlement_revision: int,
+    ) -> TransferFailureDiscardResult:
+        """原子删除指定 FAILED pending、步骤证据并解除历史回执映射。"""
+        with self._session_factory() as session:
+            transaction = SqlAlchemyUnitOfWork(session)
+            try:
+                pending_oper = TransferPendingOper(session)
+                history_oper = TransferHistoryOper(session)
+                pending = pending_oper.get_by_task_id(task_id=task_id)
+                if pending is None:
+                    history = history_oper.get(history_id)
+                    if history is None or history.transfer_task_id is None:
+                        return TransferFailureDiscardResult(
+                            discarded=True,
+                            state=None,
+                            message="失败整理任务已由并发请求放弃",
+                        )
+                    return TransferFailureDiscardResult(
+                        discarded=False,
+                        state=None,
+                        message="未找到与失败历史匹配的整理任务，请刷新后重试",
+                    )
+
+                state = TransferExecutionState(pending.execution_state)
+                if state is not TransferExecutionState.FAILED:
+                    message = (
+                        "人工复核任务必须先完成专门判定"
+                        if state is TransferExecutionState.MANUAL_REVIEW
+                        else "整理任务当前状态不能放弃"
+                    )
+                    return TransferFailureDiscardResult(
+                        discarded=False,
+                        state=state,
+                        message=message,
+                    )
+                if pending.lease_owner is not None or pending.lease_token is not None:
+                    return TransferFailureDiscardResult(
+                        discarded=False,
+                        state=state,
+                        message="整理任务仍被执行器占用，不能放弃",
+                    )
+                if (
+                        pending.terminal_history_id != history_id
+                        or pending.settlement_revision != settlement_revision
+                ):
+                    return TransferFailureDiscardResult(
+                        discarded=False,
+                        state=state,
+                        message="整理任务失败回执已变化，请刷新后重试",
+                    )
+
+                deleted = pending_oper.stage_delete_terminal_failure(
+                    task_id=task_id,
+                    history_id=history_id,
+                    settlement_revision=settlement_revision,
+                )
+                if deleted != 1:
+                    session.expire_all()
+                    pending = pending_oper.get_by_task_id(task_id=task_id)
+                    history = history_oper.get(history_id)
+                    if pending is None and (
+                            history is None or history.transfer_task_id is None
+                    ):
+                        return TransferFailureDiscardResult(
+                            discarded=True,
+                            state=None,
+                            message="失败整理任务已由并发请求放弃",
+                        )
+                    return TransferFailureDiscardResult(
+                        discarded=False,
+                        state=(
+                            TransferExecutionState(pending.execution_state)
+                            if pending is not None
+                            else None
+                        ),
+                        message="整理任务状态已变化，未放弃失败任务",
+                    )
+
+                # PostgreSQL/启用外键的 SQLite 会级联删除；显式清理兼容独立测试库。
+                TransferExecutionStepOper(session).stage_delete_task(task_id=task_id)
+                detached = history_oper.stage_detach_failed_transfer_task(
+                    history_id=history_id,
+                    task_id=task_id,
+                    settlement_revision=settlement_revision,
+                )
+                if detached != 1:
+                    raise TransferExecutionConflictError(
+                        "失败整理任务删除后无法解除历史回执映射"
+                    )
+                transaction.commit()
+                return TransferFailureDiscardResult(
+                    discarded=True,
+                    state=TransferExecutionState.FAILED,
+                    message="已放弃失败整理任务",
+                )
+            except Exception:
+                self._rollback(transaction)
+                raise
+
    def resolve_manual_review(
            self,
            *,
--- a/app/db/models/transferhistory.py
+++ b/app/db/models/transferhistory.py
@@ -627,6 +627,41 @@ class TransferHistory(Base):
            update(cls).where(cls.id == historyid).values(download_hash=download_hash)
        )

+    @classmethod
+    def detach_failed_transfer_task(
+            cls,
+            db: Session,
+            *,
+            history_id: int,
+            task_id: str,
+            settlement_revision: int,
+    ) -> int:
+        """
+        解除指定失败回执与 durable 任务的精确映射。
+
+        :param db: 数据库会话
+        :param history_id: 失败终态历史标识
+        :param task_id: 稳定任务标识
+        :param settlement_revision: 失败回执结算版本
+        :return: 更新的历史数
+        """
+        if history_id <= 0 or not task_id or settlement_revision <= 0:
+            return 0
+        return execute_dml(
+            db,
+            update(cls)
+            .where(
+                cls.id == history_id,
+                cls.transfer_task_id == task_id,
+                cls.transfer_settlement_revision == settlement_revision,
+            )
+            .values(
+                transfer_task_id=None,
+                transfer_settlement_revision=None,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+
    @classmethod
    def replace_by_src(cls, db: Session, **kwargs) -> "TransferHistory":
        """
--- a/app/db/models/transferpending.py
+++ b/app/db/models/transferpending.py
@@ -27,7 +27,12 @@ from sqlalchemy.sql.elements import ColumnElement
 from app.db.base import Base, execute_dml, get_id_column

 _TRANSFER_EXECUTION_STEP = table("transferexecutionstep", column("task_id"))
-_TRANSFER_HISTORY = table("transferhistory", column("transfer_task_id"))
+_TRANSFER_HISTORY = table(
+    "transferhistory",
+    column("id"),
+    column("transfer_task_id"),
+    column("transfer_settlement_revision"),
+)


 class TransferPending(Base):
@@ -737,6 +742,49 @@ class TransferPending(Base):
            execution_options={"synchronize_session": False},
        )

+    @classmethod
+    def delete_terminal_failure(
+            cls,
+            db: Session,
+            *,
+            task_id: str,
+            history_id: int,
+            settlement_revision: int,
+    ) -> int:
+        """
+        仅删除仍与指定失败历史精确绑定的无租约终态任务。
+
+        历史存在性子查询把用户看到的失败回执纳入同一个 CAS，避免陈旧页面误删
+        已重试、已人工判定或已产生新结算版本的任务。
+        :param db: 数据库会话
+        :param task_id: 稳定任务标识
+        :param history_id: 失败终态历史标识
+        :param settlement_revision: 失败回执结算版本
+        :return: 删除的任务数，1 表示成功取得放弃权
+        """
+        if not task_id or history_id <= 0 or settlement_revision <= 0:
+            return 0
+        return execute_dml(
+            db,
+            delete(cls).where(
+                cls.task_id == task_id,
+                cls.execution_state == "failed",
+                cls.terminal_history_id == history_id,
+                cls.settlement_revision == settlement_revision,
+                cls.lease_owner.is_(None),
+                cls.lease_token.is_(None),
+                exists(
+                    select(_TRANSFER_HISTORY.c.id).where(
+                        _TRANSFER_HISTORY.c.id == history_id,
+                        _TRANSFER_HISTORY.c.transfer_task_id == cls.task_id,
+                        _TRANSFER_HISTORY.c.transfer_settlement_revision
+                        == cls.settlement_revision,
+                    )
+                ),
+            ),
+            execution_options={"synchronize_session": False},
+        )
+
    @classmethod
    def delete_terminal_success(
            cls,
--- a/app/db/oper/transferhistory.py
+++ b/app/db/oper/transferhistory.py
@@ -509,6 +509,23 @@ class TransferHistoryOper(DbOper):
        self._db.flush()
        return history

+    def stage_detach_failed_transfer_task(
+            self,
+            *,
+            history_id: int,
+            task_id: str,
+            settlement_revision: int,
+    ) -> int:
+        """在调用方事务内解除指定失败历史与 durable 任务的精确映射。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("整理失败回执解绑需要调用方提供同步 Session")
+        return TransferHistory.detach_failed_transfer_task(
+            self._db,
+            history_id=history_id,
+            task_id=task_id,
+            settlement_revision=settlement_revision,
+        )
+
    def update_download_hash(self, historyid, download_hash):
        """
        补充转移记录download_hash
--- a/app/db/oper/transferpending.py
+++ b/app/db/oper/transferpending.py
@@ -521,6 +521,23 @@ class TransferPendingOper(DbOper):
            )
        )

+    def stage_delete_terminal_failure(
+            self,
+            *,
+            task_id: str,
+            history_id: int,
+            settlement_revision: int,
+    ) -> int:
+        """暂存与指定失败回执精确匹配的无租约 pending 删除。"""
+        return self._execute_sync_write(
+            lambda session: TransferPending.delete_terminal_failure(
+                session,
+                task_id=task_id,
+                history_id=history_id,
+                settlement_revision=settlement_revision,
+            )
+        )
+
    def stage_delete_terminal_success(
            self,
            *,
--- a/app/modules/filemanager/transhandler.py
+++ b/app/modules/filemanager/transhandler.py
@@ -20,6 +20,7 @@ from app.application.transfer.workflow import (
    TransferPlanCheckpoint,
    TransferPlanItem,
    TransferPlanningInput,
+    TransferPlanningRejectedError,
 )
 from app.domain.context import MediaInfo, MusicInfo
 from app.domain.meta.metabase import MetaBase
@@ -380,7 +381,7 @@ class TransHandler:
                        preview=preview,
                        skip_reason="未识别到文件集数，识别为特典/附加视频文件",
                    )
-                raise ValueError("未识别到文件集数")
+                raise TransferPlanningRejectedError("未识别到文件集数")
            planning_meta.end_season = None
            if planning_meta.total_season:
                planning_meta.total_season = 1
--- a/tests/test_chain_durable_events.py
+++ b/tests/test_chain_durable_events.py
@@ -1059,6 +1059,7 @@ def test_receipt_replay_survives_real_history_command_cleanup(cleanup):
        command = TransferHistoryMutationCommand(
            repository=TransferHistoryOper(session),
            download_repository=Mock(),
+            transfer_execution_repository=Mock(),
            unit_of_work=SqlAlchemyUnitOfWork(session),
            file_item_factory=Mock(),
            delete_media_file=Mock(return_value=True),
--- a/tests/test_history_mutation_command.py
+++ b/tests/test_history_mutation_command.py
@@ -7,6 +7,10 @@ from app.application.history import (
    DownloadHistoryMutationCommand,
    TransferHistoryMutationCommand,
 )
+from app.application.transfer.execution import (
+    TransferExecutionState,
+    TransferFailureDiscardResult,
+)


 def _transfer_command(*, history=None, delete_result=True, commit_error=None):
@@ -14,11 +18,13 @@ def _transfer_command(*, history=None, delete_result=True, commit_error=None):
    repository = Mock()
    repository.get.return_value = history
    download_repository = Mock()
+    transfer_execution_repository = Mock()
    unit_of_work = Mock()
    unit_of_work.commit.side_effect = commit_error
    dependencies = {
        "repository": repository,
        "download_repository": download_repository,
+        "transfer_execution_repository": transfer_execution_repository,
        "unit_of_work": unit_of_work,
        "file_item_factory": lambda payload: SimpleNamespace(**payload),
        "delete_media_file": Mock(return_value=delete_result),
@@ -38,6 +44,7 @@ def _history():
        src_fileitem={"path": "/downloads/demo.mkv"},
        dest_fileitem={"path": "/media/demo.mkv"},
        transfer_task_id=None,
+        transfer_settlement_revision=None,
    )


@@ -129,22 +136,57 @@ def test_transfer_truncate_uses_single_transaction():
    dependencies["unit_of_work"].commit.assert_called_once_with()


-def test_transfer_delete_rejects_durable_receipt_before_file_side_effects():
-    """durable 回执不能被历史 API 连同源或目标文件一起删除。"""
+def test_transfer_delete_rejects_nonfailed_durable_receipt_before_file_side_effects():
+    """非 FAILED durable 回执不能被历史 API 连同文件一起删除。"""
    history = _history()
    history.transfer_task_id = "task-durable"
+    history.transfer_settlement_revision = 2
    command, dependencies = _transfer_command(history=history)
+    dependencies["transfer_execution_repository"].discard_failed.return_value = (
+        TransferFailureDiscardResult(
+            discarded=False,
+            state=TransferExecutionState.MANUAL_REVIEW,
+            message="人工复核任务必须先完成专门判定",
+        )
+    )

    result = command.delete(7, delete_source=True, delete_destination=True)

    assert result.success is False
-    assert result.message == "持久整理失败记录不可删除，请使用重试或人工复核入口"
+    assert result.message == "人工复核任务必须先完成专门判定"
    assert result.history == "retained"
    dependencies["delete_media_file"].assert_not_called()
    dependencies["repository"].stage_delete.assert_not_called()
    dependencies["unit_of_work"].commit.assert_not_called()


+def test_transfer_delete_discards_failed_durable_receipt_before_cleanup():
+    """FAILED durable 回执应先放弃任务，再按普通历史执行文件和记录删除。"""
+    history = _history()
+    history.transfer_task_id = "task-durable"
+    history.transfer_settlement_revision = 3
+    command, dependencies = _transfer_command(history=history)
+    dependencies["transfer_execution_repository"].discard_failed.return_value = (
+        TransferFailureDiscardResult(
+            discarded=True,
+            state=TransferExecutionState.FAILED,
+            message="已放弃失败整理任务",
+        )
+    )
+
+    result = command.delete(7, delete_destination=True)
+
+    assert result.success is True
+    dependencies["transfer_execution_repository"].discard_failed.assert_called_once_with(
+        task_id="task-durable",
+        history_id=7,
+        settlement_revision=3,
+    )
+    dependencies["delete_media_file"].assert_called_once()
+    dependencies["repository"].stage_delete.assert_called_once_with(7)
+    dependencies["unit_of_work"].commit.assert_called_once_with()
+
+
 def test_transfer_destination_delete_failure_keeps_history_and_reports_step():
    """目标文件删除失败时必须保留历史，并向调用方报告失败步骤。"""
    command, dependencies = _transfer_command(
@@ -169,6 +211,7 @@ def test_transfer_delete_commits_completed_source_when_destination_fails():
    dependencies = {
        "repository": repository,
        "download_repository": Mock(),
+        "transfer_execution_repository": Mock(),
        "unit_of_work": Mock(),
        "file_item_factory": lambda payload: SimpleNamespace(**payload),
        "file_exists": Mock(return_value=True),
@@ -201,6 +244,9 @@ def test_transfer_delete_treats_missing_requested_file_as_completed():
    command = TransferHistoryMutationCommand(
        repository=dependencies["repository"],
        download_repository=dependencies["download_repository"],
+        transfer_execution_repository=dependencies[
+            "transfer_execution_repository"
+        ],
        unit_of_work=dependencies["unit_of_work"],
        file_item_factory=dependencies["file_item_factory"],
        file_exists=dependencies["file_exists"],
--- a/tests/test_manual_transfer_history.py
+++ b/tests/test_manual_transfer_history.py
@@ -622,6 +622,61 @@ def test_manual_reorganize_removes_success_history_and_old_target(monkeypatch):
    assert planned == [fileitem.path]


+def test_explicit_durable_reorganize_discards_old_task_and_replans(monkeypatch):
+    """显式重新整理 durable 失败历史时不得登记原计划重试，应清理后重新规划。"""
+    chain = make_transfer_chain()
+    fileitem = make_fileitem("/downloads/Test.Show.S01E01.mkv")
+    history = SimpleNamespace(
+        id=12,
+        transfer_task_id="transfer-task-12",
+        transfer_settlement_revision=1,
+        status=False,
+        mode="copy",
+        dest_fileitem=None,
+        download_hash=None,
+        downloader=None,
+        src=fileitem.path,
+        src_storage=fileitem.storage,
+    )
+    planned = []
+    deleted = []
+    _patch_transfer_planning(
+        monkeypatch,
+        chain,
+        fileitem,
+        history,
+        planned,
+        deleted,
+    )
+    monkeypatch.setattr(
+        chain,
+        "_request_durable_transfer_retry",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("显式重新整理不得登记旧 durable 计划重试")
+        ),
+    )
+    monkeypatch.setattr(
+        chain,
+        "_delete_manual_transfer_history",
+        lambda history, transfer_history_oper: deleted.append(
+            ("history", history.id)
+        ) or (True, ""),
+    )
+
+    state, message = TransferChain.do_transfer(
+        chain,
+        fileitem=fileitem,
+        background=False,
+        manual=True,
+        reorganize=True,
+    )
+
+    assert state is True
+    assert message == ""
+    assert deleted == [("history", 12)]
+    assert planned == [fileitem.path]
+
+
 def test_manual_reorganize_keeps_successful_move_target_as_source(monkeypatch):
    """成功移动后的目标是当前重整源，只能删历史记录，不能先删除文件。"""
    chain = make_transfer_chain()
--- a/tests/test_transfer_durable_retry_owner.py
+++ b/tests/test_transfer_durable_retry_owner.py
@@ -4,6 +4,7 @@ from types import SimpleNamespace

 from app.application.transfer.execution import (
    TransferExecutionState,
+    TransferFailureDiscardResult,
    TransferRetryRequestResult,
 )
 from app.chain.transfer import TransferChain
@@ -31,6 +32,26 @@ class _RetryCommand:
        return self.result


+class _DiscardCommand:
+    """记录显式重新整理提交的失败任务放弃请求。"""
+
+    calls: list[tuple[object, dict]] = []
+    result = TransferFailureDiscardResult(
+        discarded=True,
+        state=TransferExecutionState.FAILED,
+        message="已放弃失败整理任务",
+    )
+
+    def __init__(self, repository: object) -> None:
+        """保存测试仓储实例。"""
+        self._repository = repository
+
+    def discard_failed(self, **kwargs) -> TransferFailureDiscardResult:
+        """记录放弃请求并返回用例指定结果。"""
+        self.calls.append((self._repository, kwargs))
+        return self.result
+
+
 def _install_retry_port(monkeypatch) -> object:
    """构造不会接触数据库的 execution 端口并安装命令替身。"""
    repository = object()
@@ -48,6 +69,22 @@ def _install_retry_port(monkeypatch) -> object:
    return repository


+def _install_discard_port(monkeypatch) -> object:
+    """构造不会接触数据库的 execution 端口并安装放弃命令替身。"""
+    repository = object()
+    _DiscardCommand.calls = []
+    _DiscardCommand.result = TransferFailureDiscardResult(
+        discarded=True,
+        state=TransferExecutionState.FAILED,
+        message="已放弃失败整理任务",
+    )
+    monkeypatch.setattr(
+        "app.chain.transfer.records.TransferExecutionCommand",
+        _DiscardCommand,
+    )
+    return repository
+
+
 def test_durable_history_redo_only_requests_persistent_retry(monkeypatch):
    """durable 重做不得检查源文件、重新识别或重新准入执行。"""
    repository = _install_retry_port(monkeypatch)
@@ -91,12 +128,13 @@ def test_durable_history_redo_only_requests_persistent_retry(monkeypatch):
    ]


-def test_durable_manual_cleanup_keeps_target_history_and_failure_budget(monkeypatch):
-    """手动重整命中 durable 历史时不得先删目标、历史或失败计数。"""
-    repository = _install_retry_port(monkeypatch)
+def test_durable_manual_cleanup_discards_task_and_removes_old_state(monkeypatch):
+    """显式重整命中 FAILED durable 历史时应放弃任务并清理旧状态。"""
+    repository = _install_discard_port(monkeypatch)
    history = SimpleNamespace(
        id=82,
        transfer_task_id="transfer-task-82",
+        transfer_settlement_revision=4,
        status=False,
        mode="copy",
        src="/downloads/source.mkv",
@@ -107,22 +145,23 @@ def test_durable_manual_cleanup_keeps_target_history_and_failure_budget(monkeypa
            "type": "file",
        },
    )
-    history_port = SimpleNamespace(
-        delete=lambda _history_id: (_ for _ in ()).throw(
-            AssertionError("durable 历史不得删除")
-        )
-    )
+    deleted = []
+    cleared = []
+    history_port = SimpleNamespace(delete=lambda history_id: deleted.append(
+        ("history", history_id)
+    ))
    monkeypatch.setattr(
        "app.chain.transfer.records.StorageChain",
-        lambda: (_ for _ in ()).throw(
-            AssertionError("durable 目标不得删除")
+        lambda: SimpleNamespace(
+            exists=lambda _fileitem: True,
+            delete_media_file=lambda fileitem: deleted.append(
+                ("target", fileitem.path)
+            ) or True,
        ),
    )
    monkeypatch.setattr(
        "app.chain.transfer.records.clear_transfer_failures",
-        lambda *_args: (_ for _ in ()).throw(
-            AssertionError("durable 失败计数不得清零")
-        ),
+        lambda *args: cleared.append(args),
    )
    chain = object.__new__(TransferChain)
    chain.transfer_execution_repository = repository
@@ -132,9 +171,72 @@ def test_durable_manual_cleanup_keeps_target_history_and_failure_budget(monkeypa
        transfer_history_oper=history_port,
    )

+    assert state is True
+    assert message == ""
+    assert _DiscardCommand.calls == [
+        (
+            repository,
+            {
+                "task_id": "transfer-task-82",
+                "history_id": 82,
+                "settlement_revision": 4,
+            },
+        )
+    ]
+    assert deleted == [
+        ("target", "/library/source.mkv"),
+        ("history", 82),
+    ]
+    assert cleared == [("/downloads/source.mkv", "local")]
+
+
+def test_durable_manual_cleanup_rejects_nonfailed_state(monkeypatch):
+    """非 FAILED durable 任务被拒绝后不得删除目标、历史或失败计数。"""
+    repository = _install_discard_port(monkeypatch)
+    _DiscardCommand.result = TransferFailureDiscardResult(
+        discarded=False,
+        state=TransferExecutionState.MANUAL_REVIEW,
+        message="人工复核任务必须先完成专门判定",
+    )
+    history = SimpleNamespace(
+        id=85,
+        transfer_task_id="transfer-task-85",
+        transfer_settlement_revision=2,
+        status=False,
+        mode="copy",
+        src="/downloads/source.mkv",
+        src_storage="local",
+        dest_fileitem={
+            "storage": "local",
+            "path": "/library/source.mkv",
+            "type": "file",
+        },
+    )
+    monkeypatch.setattr(
+        "app.chain.transfer.records.StorageChain",
+        lambda: (_ for _ in ()).throw(AssertionError("拒绝后不得删除目标")),
+    )
+    monkeypatch.setattr(
+        "app.chain.transfer.records.clear_transfer_failures",
+        lambda *_args: (_ for _ in ()).throw(
+            AssertionError("拒绝后不得清理失败计数")
+        ),
+    )
+    chain = object.__new__(TransferChain)
+    chain.transfer_execution_repository = repository
+    history_port = SimpleNamespace(
+        delete=lambda _history_id: (_ for _ in ()).throw(
+            AssertionError("拒绝后不得删除历史")
+        )
+    )
+
+    state, message = chain._delete_manual_transfer_history(
+        history=history,
+        transfer_history_oper=history_port,
+    )
+
    assert state is False
-    assert message == "整理任务已登记重试"
-    assert _RetryCommand.calls[0][1]["requested_by"] == "manual_reorganize"
+    assert message == "人工复核任务必须先完成专门判定"


 def test_durable_ai_button_bypasses_agent_and_requests_scheduler(monkeypatch):
--- a/tests/test_transfer_execution_persistence.py
+++ b/tests/test_transfer_execution_persistence.py
@@ -134,6 +134,56 @@ def _seed_pending(
        session.commit()


+def _seed_failed_receipt(
+        factory,
+        *,
+        task_id: str = "task-1",
+        execution_state: str = "failed",
+        settlement_revision: int = 1,
+) -> int:
+    """写入带失败历史和步骤证据的无租约 durable 终态任务。"""
+    _seed_pending(factory, task_id=task_id)
+    with factory() as session:
+        pending = session.scalar(
+            select(TransferPending).where(TransferPending.task_id == task_id)
+        )
+        assert pending is not None
+        history = TransferHistory(
+            transfer_task_id=task_id,
+            transfer_settlement_revision=settlement_revision,
+            src=f"/{task_id}.mkv",
+            src_storage="local",
+            status=False,
+            errmsg="terminal failure",
+        )
+        session.add(history)
+        session.flush()
+        pending.execution_state = execution_state
+        pending.settlement_revision = settlement_revision
+        pending.terminal_history_id = history.id
+        pending.lease_owner = None
+        pending.lease_token = None
+        pending.lease_expires_at = None
+        session.add(TransferExecutionStep(
+            task_id=task_id,
+            operation_id=f"operation-{task_id}",
+            checkpoint_fingerprint=_plan_fingerprint(task_id),
+            ordinal=0,
+            phase="transfer",
+            kind="materialize_target",
+            state="failed",
+            attempt_token=None,
+            attempt_count=1,
+            intent_version=1,
+            intent_payload={"path": f"/{task_id}.mkv"},
+            last_error="terminal failure",
+            prepared_at="2026-08-27 09:00:00",
+            updated_at="2026-08-27 09:30:00",
+        ))
+        session.commit()
+        return history.id
+
+
 def _repository(factory, token_values: list[str] | None = None):
    """构造固定时钟与可预测 attempt token 的执行命令。"""
    repository = TransactionalTransferExecutionRepository(
@@ -833,6 +883,81 @@ def test_user_retry_is_single_generation_and_rejects_manual_review(execution_sto
    assert "人工" in rejected.message


+def test_discard_failed_removes_execution_evidence_and_detaches_history(
+        execution_store,
+) -> None:
+    """放弃匹配的 FAILED 任务应删除执行证据，并把历史恢复为普通记录。"""
+    history_id = _seed_failed_receipt(execution_store)
+    _, command = _repository(execution_store)
+
+    result = command.discard_failed(
+        task_id="task-1",
+        history_id=history_id,
+        settlement_revision=1,
+    )
+
+    assert result.discarded is True
+    assert result.state is TransferExecutionState.FAILED
+    with execution_store() as session:
+        assert session.scalar(select(TransferPending)) is None
+        assert session.scalar(select(TransferExecutionStep)) is None
+        history = session.scalar(select(TransferHistory))
+        assert history is not None
+        assert history.id == history_id
+        assert history.transfer_task_id is None
+        assert history.transfer_settlement_revision is None
+
+
+@pytest.mark.parametrize("execution_state", ("retry_wait", "manual_review"))
+def test_discard_failed_rejects_nonfailed_execution_state(
+        execution_store,
+        execution_state,
+) -> None:
+    """等待重试和人工复核任务不得通过历史维护入口删除执行证据。"""
+    history_id = _seed_failed_receipt(
+        execution_store,
+        execution_state=execution_state,
+    )
+    _, command = _repository(execution_store)
+
+    result = command.discard_failed(
+        task_id="task-1",
+        history_id=history_id,
+        settlement_revision=1,
+    )
+
+    assert result.discarded is False
+    assert result.state is TransferExecutionState(execution_state)
+    with execution_store() as session:
+        assert session.scalar(select(TransferPending)) is not None
+        assert session.scalar(select(TransferExecutionStep)) is not None
+        history = session.scalar(select(TransferHistory))
+        assert history is not None
+        assert history.transfer_task_id == "task-1"
+
+
+def test_discard_failed_rejects_stale_settlement_revision(execution_store) -> None:
+    """陈旧页面携带的结算版本不能放弃已经变化的失败任务。"""
+    history_id = _seed_failed_receipt(execution_store, settlement_revision=3)
+    _, command = _repository(execution_store)
+
+    result = command.discard_failed(
+        task_id="task-1",
+        history_id=history_id,
+        settlement_revision=2,
+    )
+
+    assert result.discarded is False
+    assert "已变化" in result.message
+    with execution_store() as session:
+        pending = session.scalar(select(TransferPending))
+        history = session.scalar(select(TransferHistory))
+        assert pending is not None
+        assert pending.settlement_revision == 3
+        assert history is not None
+        assert history.transfer_settlement_revision == 3
+
+
 def test_manual_not_applied_decision_is_audited_and_schedules_same_step(
        execution_store,
 ) -> None:
--- a/tests/test_transfer_failed_retry_budget.py
+++ b/tests/test_transfer_failed_retry_budget.py
@@ -10,7 +10,6 @@ from types import SimpleNamespace
 from unittest.mock import Mock

 from app import schemas
-from app.runtime.config import settings
 from app.application.history import (
    HistoryGateAction,
    TransferHistoryMutationCommand,
@@ -19,6 +18,7 @@ from app.application.history import (
    failed_retry_count,
    record_transfer_failure,
 )
+from app.runtime.config import settings


 def _reset_failed_retries(src_path, storage=None):
@@ -115,6 +115,7 @@ def test_delete_transfer_history_endpoint_clears_retry_count(monkeypatch):
    command = TransferHistoryMutationCommand(
        repository=repository,
        download_repository=Mock(),
+        transfer_execution_repository=Mock(),
        unit_of_work=Mock(),
        file_item_factory=lambda payload: SimpleNamespace(**payload),
        delete_media_file=Mock(return_value=True),
--- a/tests/test_transfer_planning_checkpoint.py
+++ b/tests/test_transfer_planning_checkpoint.py
@@ -1344,6 +1344,33 @@ def test_checkpoint_commit_failure_blocks_executor_and_keeps_accepted():
    assert "commit failed" in repository.record_planning_failure.call_args.kwargs["error"]


+def test_host_planning_value_error_commits_rejection_instead_of_retrying():
+    """确定性的宿主规划错误必须形成失败终态，不能留在 accepted 周期回放。"""
+    repository = Mock()
+    task = _task()
+    planning_input = _planning_input()
+    task.bind_admission_task_id("task-planning-rejection")
+    _bind_planning_input(task, planning_input)
+    repository.checkpoint_plan.side_effect = lambda **kwargs: _planned_admission(
+        task,
+        kwargs["checkpoint"],
+    )
+    chain = _chain(repository=repository)
+    chain.plan_transfer.side_effect = transfer_application.TransferPlanningRejectedError(
+        "未识别到文件集数"
+    )
+
+    result = chain._plan_checkpoint_and_execute(task)
+
+    assert result.success is False
+    assert result.message == "未识别到文件集数"
+    assert task.plan_checkpoint is not None
+    assert task.plan_checkpoint.rejection_error == "未识别到文件集数"
+    assert task.plan_checkpoint.items == ()
+    assert task.execution_checkpoint is not None
+    repository.record_planning_failure.assert_not_called()
+
+
 def test_post_commit_crash_replays_frozen_plan_without_replanning():
    """commit 后执行崩溃仍保留 planned checkpoint，重启后跳过 rename 规划。"""
    repository = Mock()
--- a/tests/test_transfer_planning_compat.py
+++ b/tests/test_transfer_planning_compat.py
@@ -205,7 +205,7 @@ def test_chain_base_plan_transfer_keeps_internal_dto_type_only():
    """规划入口运行时只做内部调度，不要求导入或重复导出 DTO。"""
    chain = object.__new__(ChainBase)
    checkpoint = Mock(name="checkpoint")
-    chain.run_module = Mock(return_value=checkpoint)
+    chain.run_module_strict = Mock(return_value=checkpoint)
    meta = Mock(name="meta")
    mediainfo = Mock(name="mediainfo")

@@ -216,7 +216,7 @@ def test_chain_base_plan_transfer_keeps_internal_dto_type_only():
    )

    assert returned is checkpoint
-    chain.run_module.assert_called_once_with(
+    chain.run_module_strict.assert_called_once_with(
        "plan_transfer",
        fileitem=_fileitem(),
        meta=meta,