fix(transfer): report durable submission outcomes per file

app/api/endpoints/transfer.py

@@ -42,6 +42,8 @@ from app.schemas.transfer import EpisodeFormatRecommendData as _SchemaEpisodeFor
 from app.schemas.transfer import EpisodeFormatRecommendItem, ManualTransferItem
 from app.schemas.transfer import ManualTransferHistoryInfo as _SchemaManualTransferHistoryInfo
 from app.schemas.transfer import ManualTransferResultData as _SchemaManualTransferResultData
+from app.schemas.transfer import ManualTransferSubmissionData as _SchemaManualTransferSubmissionData
+from app.schemas.transfer import ManualTransferSubmissionItem as _SchemaManualTransferSubmissionItem
 from app.schemas.transfer import ManualTransferTargetPath as _SchemaManualTransferTargetPath
 from app.schemas.transfer import TransferJob as _SchemaTransferJob
 from app.schemas.transfer import TransferManualReviewData as _SchemaTransferManualReviewData
@@ -87,8 +89,8 @@ def _public_transfer_result(data: dict[str, Any]) -> dict[str, Any]:
                     public_item.get("message"),
                     overwrite_skipped=bool(public_item.get("overwrite_skipped")),
                 )
-                public_item.setdefault("failure_stage", feedback.stage.value)
-                public_item.setdefault("recovery_action", feedback.action)
+                public_item["failure_stage"] = public_item.get("failure_stage") or feedback.stage.value
+                public_item["recovery_action"] = public_item.get("recovery_action") or feedback.action
             public_items.append(public_item)
         result["items"] = public_items
     return result
@@ -154,6 +156,41 @@ def _merge_transfer_messages(messages: List[str]) -> str:
     )
 
 
+def _manual_submission_items(
+    fileitem: FileItem, success: bool, result: object, target_path: Optional[Path],
+) -> list[_SchemaManualTransferSubmissionItem]:
+    """映射真实执行回执，旧返回值最多确认接收，不能推断入库完成。"""
+    if isinstance(result, dict):
+        items = _public_transfer_result(result).get("items")
+        if isinstance(items, list):
+            return [_SchemaManualTransferSubmissionItem.model_validate(item) for item in items]
+    message = _public_transfer_message(result)
+    data = _public_transfer_result({"items": [{
+        "source": fileitem.path, "success": success,
+        "state": "accepted" if success else "failed", "message": message,
+        "target_dir": target_path.as_posix() if target_path else None,
+    }]})
+    return [_SchemaManualTransferSubmissionItem.model_validate(item) for item in data["items"]]
+
+
+def _manual_submission_response(
+    success: bool, items: list[_SchemaManualTransferSubmissionItem],
+) -> Any:
+    """保留部分成功文件，并继续为旧客户端提供包含失败路径的总提示。"""
+    errors = [
+        _format_manual_transfer_failure(
+            message=item.message, source_path=item.source,
+            target_path=item.target or item.target_dir,
+        )
+        for item in items if item.state in {"failed", "retry_wait", "manual_review"} and item.message
+    ]
+    message = _merge_transfer_messages(errors or [item.message for item in items if item.message and not success])
+    return _SchemaResponse(
+        success=success, message=message,
+        data=_SchemaManualTransferSubmissionData(items=items, message=message),
+    )
+
+
 def _is_music_file_batch(fileitems: List[FileItem]) -> bool:
     """判断显式选中项是否为可按专辑识别的多音轨批次。"""
     if len(fileitems) < 2 or any(fileitem.type != "file" for fileitem in fileitems):
@@ -608,7 +645,7 @@ def query_manual_transfer_history(
 @router.post(
     "/manual",
     summary="手动转移",
-    response_model=_SchemaResponse[_SchemaManualTransferResultData],
+    response_model=_SchemaResponse[_SchemaManualTransferSubmissionData | _SchemaManualTransferResultData],
 )
 def manual_transfer(
     transer_item: ManualTransferItem,
@@ -781,6 +818,7 @@ def _execute_manual_transfer(
     # 前端显式传入文件列表时，按选中的文件逐个处理，避免将目录整体展开。
     if explicit_selected_files:
         preview_items: List[dict] = []
+        submission_items: list[_SchemaManualTransferSubmissionItem] = []
         error_messages: List[str] = []
         all_success = True
         for src_fileitem in src_fileitems:
@@ -810,6 +848,7 @@ def _execute_manual_transfer(
                 reorganize=transer_item.reorganize,
                 sync_extra_files=False,
                 cleanup_dest_fileitem=cleanup_dest_fileitem,
+                report_results=not transer_item.preview,
             )
             if transer_item.preview:
                 if isinstance(errormsg, dict):
@@ -832,19 +871,9 @@ def _execute_manual_transfer(
                         )
                     )
                     all_success = False
-            elif not state:
-                all_success = False
-                if isinstance(errormsg, list):
-                    failure_message = "；".join(str(msg) for msg in errormsg if msg)
-                else:
-                    failure_message = str(errormsg or "整理失败")
-                error_messages.append(
-                    _format_manual_transfer_failure(
-                        message=failure_message,
-                        source_path=src_fileitem.path,
-                        target_path=target_path.as_posix() if target_path else None,
-                    )
-                )
+            else:
+                all_success = all_success and state
+                submission_items.extend(_manual_submission_items(src_fileitem, state, errormsg, target_path))
 
         if transer_item.preview:
             merged_preview_items: List[dict] = []
@@ -879,12 +908,7 @@ def _execute_manual_transfer(
                 data=preview_data,
             )
 
-        if not all_success:
-            return _SchemaResponse(
-                success=False,
-                message=_merge_transfer_messages(error_messages),
-            )
-        return _SchemaResponse(success=True)
+        return _manual_submission_response(all_success, submission_items)
 
     src_fileitem = src_fileitems[0]
     # 开始转移
@@ -919,30 +943,20 @@ def _execute_manual_transfer(
         sync_extra_files=selected_music_fileitems is None,
         cleanup_dest_fileitem=cleanup_dest_fileitem,
         selected_fileitems=selected_music_fileitems,
+        report_results=not transer_item.preview,
     )
-    # 失败
-    if not state:
-        if isinstance(errormsg, list):
-            errormsg = f"整理完成，{len(errormsg)} 个文件转移失败！"
-        if isinstance(errormsg, dict):
-            public_result = _public_transfer_result(errormsg)
-            return _SchemaResponse(
-                success=True,
-                message=public_result.get("message"),
-                data=public_result,
-            )
-        return _SchemaResponse(
-            success=False,
-            message=_format_manual_transfer_failure(
-                message=errormsg,
-                source_path=src_fileitem.path,
-                target_path=target_path.as_posix() if target_path else None,
-            ),
-        )
-    # 成功
-    if transer_item.preview:
+    if not transer_item.preview:
+        return _manual_submission_response(state, _manual_submission_items(src_fileitem, state, errormsg, target_path))
+    # 预览失败仍返回完整预览列表，实际提交的总状态已在上方独立处理。
+    if isinstance(errormsg, dict):
+        public_result = _public_transfer_result(errormsg)
+        return _SchemaResponse(success=True, message=public_result.get("message"), data=public_result)
+    if state:
         return _SchemaResponse(success=True, data=errormsg or {})
-    return _SchemaResponse(success=True)
+    return _SchemaResponse(success=False, message=_format_manual_transfer_failure(
+        message=errormsg, source_path=src_fileitem.path,
+        target_path=target_path.as_posix() if target_path else None,
+    ))
 
 
 @router.post(

app/chain/transfer/history.py

@@ -58,14 +58,16 @@ class TransferHistoryOwner(_TransferOwnerBase):
             self,
             transfer_kwargs: dict[str, Any],
             selected_fileitems: Optional[list[FileItem]],
+            report_results: bool = False,
     ) -> Tuple[bool, Union[str, dict[str, Any]]]:
         """显式文件批次走内部入口，普通请求继续保持公开签名兼容。"""
-        if selected_fileitems is not None:
+        if selected_fileitems is not None or report_results:
             return cast(
                 Tuple[bool, Union[str, dict[str, Any]]],
                 self._execute_transfer(
                     **transfer_kwargs,
                     selected_fileitems=selected_fileitems,
+                    report_results=report_results,
                 ),
             )
         return cast(
@@ -191,6 +193,7 @@ class TransferHistoryOwner(_TransferOwnerBase):
             music_release_regions: Optional[list[str]] = None,
             music_release_scripts: Optional[list[str]] = None,
             selected_fileitems: Optional[list[FileItem]] = None,
+            report_results: bool = False,
     ) -> Tuple[bool, Union[str, dict[str, Any]]]:
         """
         手动整理，支持复杂条件，带进度显示
@@ -220,6 +223,7 @@ class TransferHistoryOwner(_TransferOwnerBase):
         :param music_release_regions: 本次音乐整理的发行地区优先级，空值继承系统设置
         :param music_release_scripts: 本次音乐整理的文字字形优先级，空值继承系统设置
         :param selected_fileitems: 前端显式选中的批量文件
+        :param report_results: 返回实际阶段回执，后台接收不表示入库完成
         """
         logger.info(f"手动整理：{fileitem.path} ...")
         explicit_identity = media_source is not None or media_id is not None
@@ -274,12 +278,12 @@ class TransferHistoryOwner(_TransferOwnerBase):
                 music_release_regions=music_release_regions,
                 music_release_scripts=music_release_scripts,
             )
-            state, errmsg = self._run_manual_transfer_request(transfer_kwargs, selected_fileitems)
+            state, errmsg = self._run_manual_transfer_request(transfer_kwargs, selected_fileitems, report_results)
             if not state:
                 return False, errmsg
 
-            logger.info(f"{fileitem.path} 整理完成")
-            return True, errmsg if preview else ""
+            logger.info(f"{fileitem.path} 整理请求处理完成")
+            return True, errmsg if preview or report_results else ""
         else:
             # 没有输入媒体ID时，按文件识别
             transfer_kwargs = dict(
@@ -307,7 +311,7 @@ class TransferHistoryOwner(_TransferOwnerBase):
                 music_release_regions=music_release_regions,
                 music_release_scripts=music_release_scripts,
             )
-            state, errmsg = self._run_manual_transfer_request(transfer_kwargs, selected_fileitems)
+            state, errmsg = self._run_manual_transfer_request(transfer_kwargs, selected_fileitems, report_results)
             return state, errmsg
 
     def send_transfer_message(

app/chain/transfer/request.py

@@ -2,7 +2,7 @@
 
 from copy import deepcopy
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union
 
 from app.application.formatting import FormatParser
 from app.application.history import (
@@ -509,3 +509,96 @@ def build_transfer_preview_item(task: TransferTask, transferinfo: TransferInfo)
             "customization": item_meta.customization if item_meta else None,
         }
     )
+
+
+class _TransferSubmissionCollector:
+    """按当前请求收集真实阶段回执，不改变任务、重试或结算的所有权。"""
+
+    def __init__(self, enabled: bool = False, source: Optional[FileItem] = None) -> None:
+        """普通整理调用默认关闭回执，维持原有字符串返回契约。"""
+        self.enabled = enabled
+        self.source = source
+        self.items: list[dict[str, Any]] = []
+        self._results: dict[int, TransferInfo] = {}
+        self._pending: dict[tuple[Optional[str], Optional[str]], FileItem] = {}
+
+    def expect(self, fileitems: list[tuple[FileItem, bool]]) -> None:
+        """保留候选快照，取消发生后仍能说明哪些文件没有被执行。"""
+        if self.enabled:
+            self._pending.update({(item.storage, item.path): item for item, _bluray in fileitems})
+
+    def record(
+        self, fileitem: FileItem,
+        state: Literal["accepted", "completed", "failed", "retry_wait", "skipped", "manual_review"],
+        message: Optional[str] = None,
+        *, target_dir: Optional[Path] = None,
+    ) -> None:
+        """在准入、历史跳过或重试决策发生时记录回执。"""
+        if not self.enabled:
+            return
+        self._pending.pop((fileitem.storage, fileitem.path), None)
+        self.items.append({
+            "source": fileitem.path, "state": state,
+            "success": state in {"accepted", "completed", "retry_wait"},
+            "message": message,
+            "target_dir": target_dir.as_posix() if target_dir else None,
+        })
+
+    def capture(self, task: TransferTask, transferinfo: TransferInfo) -> None:
+        """暂存执行结果，等终态原子结算确认后再公布 completed。"""
+        if self.enabled:
+            self._results[id(task)] = transferinfo
+
+    def execution(
+        self, task: TransferTask, success: bool, message: str,
+        *, durable_state: Optional[str],
+    ) -> None:
+        """只有已确认的终态可报告完成，未确认终态保留后台重试状态。"""
+        if not self.enabled:
+            return
+        info = self._results.pop(id(task), None)
+        state: Literal["accepted", "completed", "failed", "retry_wait", "skipped", "manual_review"] = "failed"
+        if task.terminal_settled:
+            state = "completed" if (info.success if info else success) else "failed"
+            if info and info.overwrite_skipped:
+                state = "skipped"
+        elif durable_state == "manual_review":
+            state = "manual_review"
+        elif durable_state in {"not_started", "running", "retry_wait", "settling"}:
+            state = "retry_wait"
+        elif durable_state == "unknown":
+            state = "accepted"
+            message = f"{message}；任务状态暂未确认，请刷新整理历史"
+        elif success:
+            state = "accepted"
+        if state == "completed" and not success:
+            message = "文件已完成整理，但后续处理失败，请检查整理日志，无需重复整理"
+        self.record(task.fileitem, state, message, target_dir=task.target_path)
+        if info:
+            projected = build_transfer_preview_item(task, info)
+            self.items[-1].update({key: projected[key] for key in (
+                "target", "target_dir", "failure_stage", "recovery_action", "overwrite_skipped",
+            )})
+        if state == "manual_review":
+            self.items[-1]["recovery_action"] = "打开整理队列，先完成人工复核，再决定是否重试"
+
+    def result(
+        self, success: bool, errors: list[str], *, preview: bool,
+        preview_items: list[dict[str, Any]],
+    ) -> tuple[bool, Union[str, dict[str, Any]]]:
+        """保持预览与旧调用返回值，同时让管理界面收到完整执行回执。"""
+        message = "、".join(errors[:2]) + (f"，等{len(errors)}个文件错误！" if len(errors) > 2 else "")
+        if preview:
+            return success, {
+                "summary": {
+                    "total": len(preview_items),
+                    "success": sum(bool(item.get("success")) for item in preview_items),
+                    "failed": sum(not item.get("success") for item in preview_items),
+                },
+                "items": preview_items, "message": message,
+            }
+        for pending in list(self._pending.values()):
+            self.record(pending, "skipped" if success else "failed", message or "本次整理未执行")
+        if self.enabled and not self.items and self.source:
+            self.record(self.source, "skipped" if success else "failed", message or "没有需要执行的整理任务")
+        return success, {"items": self.items, "message": message} if self.enabled else message

app/chain/transfer/workflow.py

@@ -48,6 +48,7 @@ from app.schemas.workflow import FileItem
 from .request import (
     _should_discard_batch_music_identity,
     _TransferCandidatePlanner,
+    _TransferSubmissionCollector,
     build_transfer_preview_item,
 )
 
@@ -451,6 +452,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         music_release_regions: Optional[list[str]] = None,
         music_release_scripts: Optional[list[str]] = None,
         selected_fileitems: Optional[list[FileItem]] = None,
+        report_results: bool = False,
     ) -> Tuple[bool, Union[str, dict]]:
         """
         执行一个复杂目录的整理操作
@@ -503,6 +505,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
             return False, identity_error
 
         all_success = True
+        submission = _TransferSubmissionCollector(report_results, fileitem)
         transfer_batch_id = str(uuid.uuid4())
         batch_mtype = getattr(mediainfo, "type", None)
         if batch_mtype in (None, MediaType.UNKNOWN):
@@ -553,7 +556,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                 selected_fileitems=selected_fileitems,
             )
         except OperationInterrupted:
-            return False, f"{fileitem.name} 已取消"
+            return submission.result(False, [f"{fileitem.name} 已取消"], preview=False, preview_items=[])
 
         if not file_items:
             if has_episode_format_template and not matched_episode_format_template:
@@ -564,7 +567,8 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                         "items": [],
                         "message": "",
                     }
-                return True, ""
+                submission.record(fileitem, "skipped", "未匹配到集数定位模板，跳过整理")
+                return submission.result(True, [], preview=False, preview_items=[])
             logger.warn(f"{fileitem.path} 没有找到可整理的媒体文件")
             return False, f"{fileitem.name} 没有找到可整理的媒体文件"
 
@@ -583,6 +587,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         else:
             logger.info(f"正在计划整理 {planned_file_count} 个文件...")
 
+        submission.expect(file_items)
         try:
             (
                 transfer_tasks,
@@ -620,9 +625,10 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                 music_release_regions=music_release_regions,
                 music_release_scripts=music_release_scripts,
                 selected_music_track_map=selected_music_track_map,
+                submission=submission,
             )
         except OperationInterrupted:
-            return False, f"{fileitem.name} 已取消"
+            return submission.result(False, [f"{fileitem.name} 已取消"], preview=False, preview_items=[])
 
         all_success, err_msgs, preview_items = self._execute_transfer_tasks(
             transfer_tasks=transfer_tasks,
@@ -630,6 +636,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
             continue_callback=continue_callback,
             all_success=all_success,
             err_msgs=err_msgs,
+            submission=submission,
         )
 
         # 下载器任务在这一轮可能因为历史记录全部命中而没有进入整理队列，
@@ -639,18 +646,20 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                 logger.info(f"补充设置下载任务已整理标签：{skipped_hash}")
                 self._TransferChain__mark_torrent_completed_if_done(skipped_hash, skipped_downloader)
 
-        error_msg = "、".join(err_msgs[:2]) + (f"，等{len(err_msgs)}个文件错误！" if len(err_msgs) > 2 else "")
-        if preview:
-            return all_success, {
-                "summary": {
-                    "total": len(preview_items),
-                    "success": len([item for item in preview_items if item.get("success")]),
-                    "failed": len([item for item in preview_items if not item.get("success")]),
-                },
-                "items": preview_items,
-                "message": error_msg,
-            }
-        return all_success, error_msg
+        return submission.result(all_success, err_msgs, preview=bool(preview), preview_items=preview_items)
+
+    def _enqueue_transfer_candidate(
+        self, task: TransferTask, errors: list[str], submission: _TransferSubmissionCollector,
+    ) -> Optional[bool]:
+        """在唯一队列入口捕获准入失败，给同批其余文件保留执行机会和错误回执。"""
+        try:
+            return bool(self.put_to_queue(task=task))
+        except Exception as error:
+            logger.error(f"{task.fileitem.name} 加入整理队列失败：{error}", exc_info=True)
+            message = str(error) if isinstance(error, TransferAdmissionConflictError) else "未能加入整理队列，请稍后重试"
+            errors.append(f"{task.fileitem.name} {message}")
+            submission.record(task.fileitem, "failed", message, target_dir=task.target_path)
+            return None
 
     def _build_transfer_tasks(
         self,
@@ -684,10 +693,9 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         music_release_regions: Optional[list[str]],
         music_release_scripts: Optional[list[str]],
         selected_music_track_map: dict[str, MusicInfo],
+        submission: _TransferSubmissionCollector,
     ) -> Tuple[List[TransferTask], bool, List[str], int, set[Tuple[str, str]]]:
         """从冻结候选构建任务，并集中执行历史去重与 durable 绑定。"""
-        _build_file_meta = build_file_meta
-        # 整理所有文件
         all_success = True
         transfer_tasks: List[TransferTask] = []
         err_msgs: List[str] = []
@@ -728,6 +736,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                                 )
                                 if durable_retry is not None:
                                     accepted, message = durable_retry
+                                    submission.record(file_item, "retry_wait" if accepted else "failed", message)
                                     if accepted:
                                         logger.info(message)
                                     else:
@@ -741,6 +750,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                                 transfer_history_oper=transfer_history_oper,
                             )
                             if not state:
+                                submission.record(file_item, "failed", message)
                                 all_success = False
                                 logger.error(message)
                                 err_msgs.append(message)
@@ -764,6 +774,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                                 transferd = None
 
                         if transferd:
+                            submission.record(file_item, "skipped" if transferd.status else "failed", f"{file_item.name} 已整理过")
                             skipped_history_count += 1
                             if not transferd.status:
                                 all_success = False
@@ -809,14 +820,15 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                     elif inherited_meta:
                         file_meta = deepcopy(inherited_meta)
                     else:
-                        file_meta = _build_file_meta(
+                        file_meta = build_file_meta(
                             file_path,
                             self._get_subscribe_custom_words(download_history),
                         )
                 else:
-                    file_meta = _build_file_meta(file_path, None)
+                    file_meta = build_file_meta(file_path, None)
 
                 if not file_meta:
+                    submission.record(file_item, "failed", f"{file_path.name} 无法识别有效信息")
                     all_success = False
                     logger.error(f"{file_path.name} 无法识别有效信息")
                     err_msgs.append(f"{file_path.name} 无法识别有效信息")
@@ -846,7 +858,6 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                 if not manual and task_mediainfo and self._is_movie_year_conflict(file_meta, task_mediainfo):
                     task_mediainfo = None
 
-                # 后台整理
                 transfer_task = TransferTask(
                     fileitem=file_item,
                     meta=file_meta,
@@ -891,18 +902,12 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                     if recovery_admission.checkpoint:
                         transfer_task.bind_plan_checkpoint(recovery_admission.checkpoint)
                 if background:
-                    try:
-                        queued = self.put_to_queue(task=transfer_task)
-                    except Exception as err:
+                    queued = self._enqueue_transfer_candidate(transfer_task, err_msgs, submission)
+                    if queued is None:
                         all_success = False
-                        logger.error(
-                            f"{file_path.name} 加入整理队列失败：{err}",
-                            exc_info=True,
-                        )
-                        message = str(err) if isinstance(err, TransferAdmissionConflictError) else "未能加入整理队列，请稍后重试"
-                        err_msgs.append(f"{file_path.name} {message}")
                         continue
                     if queued:
+                        submission.record(file_item, "accepted", "已添加到整理队列", target_dir=target_path)
                         if cleanup_intent:
                             cleanup_intent_assigned = True
                         logger.info(f"{file_path.name} 已添加到整理队列")
@@ -919,9 +924,11 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                     else:
                         logger.debug(f"{file_path.name} 已在整理列表中，跳过")
                 if not queued and manual:
+                    submission.record(file_item, "failed", "已在整理队列中，请等待当前任务结束后重新整理")
                     all_success = False
                     err_msgs.append(f"{file_path.name} 已在整理队列中，请等待当前任务结束后重新整理")
         except OperationInterrupted:
+            self._cancel_unexecuted_tasks(transfer_tasks)
             raise
         finally:
             file_items.clear()
@@ -936,6 +943,24 @@ class TransferWorkflowOwner(_TransferOwnerBase):
             skipped_torrents,
         )
 
+    def _cancel_unexecuted_tasks(self, tasks: list[TransferTask]) -> None:
+        """撤销本次同步请求尚未执行的内存占位，让用户之后可以重新提交这些文件。"""
+        for task in tasks:
+            self.jobview.fail_unfinished_task(task)
+            self.jobview.try_remove_job(task)
+            self._finish_scrape_batch_task(task)
+
+    def _submission_execution_state(self, task: TransferTask, enabled: bool) -> Optional[str]:
+        """失败后读取既有持久投影，避免把等待恢复或人工复核误报为可重新提交。"""
+        if not enabled or task.terminal_settled or not task.admission_task_id:
+            return None
+        try:
+            snapshot = self.transfer_execution_repository.get_snapshot(task_id=task.admission_task_id)
+            return snapshot.state.value if snapshot else None
+        except Exception as error:
+            logger.error(f"读取整理提交回执状态失败：{task.admission_task_id} - {error}")
+            return "unknown"
+
     def _execute_transfer_tasks(
         self,
         *,
@@ -944,9 +969,9 @@ class TransferWorkflowOwner(_TransferOwnerBase):
         continue_callback: Optional[Callable[[], bool]],
         all_success: bool,
         err_msgs: List[str],
+        submission: _TransferSubmissionCollector,
     ) -> Tuple[bool, List[str], List[dict[str, Any]]]:
         """同步消费已规划任务；后台任务只由 queue owner 消费。"""
-        # 实时整理
         preview_items: List[dict[str, Any]] = []
 
         def _preview_callback(task: TransferTask, transferinfo: TransferInfo) -> Tuple[bool, str]:
@@ -973,10 +998,11 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                 logger.info(__process_msg)
                 progress.update(value=0, text=__process_msg)
             try:
-                for transfer_task in transfer_tasks:
-                    if runtime_stop_state.is_system_stopped:
-                        break
-                    if continue_callback and not continue_callback():
+                for task_index, transfer_task in enumerate(transfer_tasks):
+                    if runtime_stop_state.is_system_stopped or (continue_callback and not continue_callback()):
+                        self._cancel_unexecuted_tasks(transfer_tasks[task_index:])
+                        all_success = False
+                        err_msgs.append("整理已取消，剩余文件未执行")
                         break
                     if not preview:
                         # 更新进度
@@ -1001,6 +1027,7 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                     ) -> Tuple[bool, str]:
                         """同步路径也由默认回调原子提交历史、事件与 durable 终态。"""
                         nonlocal terminal_settlement
+                        submission.capture(callback_task, transferinfo)
                         callback = _preview_callback if preview else self._TransferChain__default_callback
                         try:
                             return callback(callback_task, transferinfo)
@@ -1037,6 +1064,10 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                     if terminal and not durable_settled:
                         state = False
                         err_msg = "整理任务结果暂未确认，后台将自动重试"
+                    submission.execution(
+                        transfer_task, state, err_msg,
+                        durable_state=self._submission_execution_state(transfer_task, submission.enabled),
+                    )
                     if not state:
                         all_success = False
                         logger.warn(f"{transfer_task.fileitem.name} {err_msg}")
@@ -1049,25 +1080,10 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                             not preview_items or preview_items[-1].get("source") != transfer_task.fileitem.path
                         ):
                             preview_items.append(
-                                {
-                                    "source": transfer_task.fileitem.path,
-                                    "target": None,
-                                    "target_dir": None,
-                                    "success": False,
-                                    "message": err_msg,
-                                    "failure_stage": "execution",
-                                    "recovery_action": "刷新整理历史，等待任务结束；仍无法继续时提交人工复核",
-                                    "type": None,
-                                    "title": None,
-                                    "season": transfer_task.meta.begin_season if transfer_task.meta else None,
-                                    "episode": transfer_task.meta.begin_episode if transfer_task.meta else None,
-                                    "episode_end": transfer_task.meta.end_episode if transfer_task.meta else None,
-                                    "part": transfer_task.meta.part if transfer_task.meta else None,
-                                    "org_string": transfer_task.meta.org_string if transfer_task.meta else None,
-                                    "apply_words": transfer_task.meta.apply_words if transfer_task.meta else [],
-                                    "resource_team": transfer_task.meta.resource_team if transfer_task.meta else None,
-                                    "customization": transfer_task.meta.customization if transfer_task.meta else None,
-                                }
+                                build_transfer_preview_item(transfer_task, TransferInfo(
+                                    success=False, message=err_msg, failure_stage="execution",
+                                    recovery_action="刷新整理历史，等待任务结束；仍无法继续时提交人工复核",
+                                ))
                             )
                         fail_num += 1
                     else:
@@ -1076,13 +1092,11 @@ class TransferWorkflowOwner(_TransferOwnerBase):
                             self.jobview.finish_task(transfer_task)
                             self.jobview.try_remove_job(transfer_task)
                         processed_num += 1
-                    # 记录已完成
                     finished_files.append(Path(transfer_task.fileitem.path).as_posix())
             finally:
                 transfer_tasks.clear()
                 del transfer_tasks
 
-            # 整理结束
             if not preview:
                 __end_msg = f"整理队列处理完成，共整理 {total_num} 个文件，失败 {fail_num} 个"
                 logger.info(__end_msg)

app/schemas/exports.py

@@ -216,6 +216,8 @@ SCHEMA_EXPORTS = {
     'ManualTransferPreviewItem': ('app.schemas.transfer', 'ManualTransferPreviewItem'),
     'ManualTransferPreviewSummary': ('app.schemas.transfer', 'ManualTransferPreviewSummary'),
     'ManualTransferResultData': ('app.schemas.transfer', 'ManualTransferResultData'),
+    'ManualTransferSubmissionData': ('app.schemas.transfer', 'ManualTransferSubmissionData'),
+    'ManualTransferSubmissionItem': ('app.schemas.transfer', 'ManualTransferSubmissionItem'),
     'ManualTransferTargetPath': ('app.schemas.transfer', 'ManualTransferTargetPath'),
     'McpJsonRpcCapabilities': ('app.schemas.mcp', 'McpJsonRpcCapabilities'),
     'McpJsonRpcClientInfo': ('app.schemas.mcp', 'McpJsonRpcClientInfo'),

app/schemas/transfer.py

@@ -400,13 +400,36 @@ class ManualTransferPreviewItem(BaseModel):
 
 
 class ManualTransferResultData(BaseModel):
-    """手动整理预览或执行结果数据。"""
+    """手动整理预览数据，不表示执行或入库已完成。"""
 
     summary: Optional[ManualTransferPreviewSummary] = None
     items: list[ManualTransferPreviewItem] = Field(default_factory=list)
     message: Optional[str] = None
 
 
+# 现有 mypy 配置把 Pydantic 基类视为 Any，公开字段仍由显式模型校验。
+class ManualTransferSubmissionItem(BaseModel):  # type: ignore[misc]
+    """实际提交的逐文件回执，接收和等待重试均不表示已入库。"""
+
+    state: Literal["accepted", "completed", "failed", "retry_wait", "skipped", "manual_review"]
+    source: Optional[str] = None
+    target: Optional[str] = None
+    target_dir: Optional[str] = None
+    # 表示本次操作被接收或完成；实际入库只以 completed 为准。
+    success: bool = False
+    message: Optional[str] = None
+    failure_stage: Optional[str] = None
+    recovery_action: Optional[str] = None
+    overwrite_skipped: bool = False
+
+
+class ManualTransferSubmissionData(BaseModel):  # type: ignore[misc]
+    """实际提交结果，即使批次失败也保留已接收、完成和跳过的文件。"""
+
+    items: list[ManualTransferSubmissionItem] = Field(default_factory=list)
+    message: Optional[str] = None
+
+
 class EpisodeFormatRecommendData(BaseModel):
     """集数定位模板推荐结果。"""
 

docs/mcp-api.md

@@ -289,6 +289,14 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返
 | GET | `/api/v1/transfer/tasks/{task_id}/manual-review` | 管理员查询单个 durable 人工复核任务详情；仅可读取 `manual_review` 或已经人工判定的 `retry_wait` 任务，其余状态按不存在处理 |
 | POST | `/api/v1/transfer/tasks/{task_id}/manual-review` | 管理员判定处于 `manual_review` 的 durable 整理步骤；请求包含 `operation_id`、`decision=not_applied|applied`、`reason`，`applied` 还必须提供 `result_payload`。`failed` 不属于公开决策，失败终态只能由持租约的 durable 结算写入；响应仅返回任务、操作、决策、后续状态和复核修订号 |
 
+`transfer/manual` 在 `preview=true` 时保留预览的 `summary/items/message`，不返回执行状态。
+实际提交返回独立的 `data.items` 回执，即使批次 `success=false` 也保留其他文件的结果。
+每项包含 `source/target/target_dir/success/message/failure_stage/recovery_action/overwrite_skipped/state`；
+`state=accepted` 仅表示已接收，`retry_wait` 表示原计划已交给后台恢复，均不代表入库。
+仅 `completed` 表示执行和终态原子结算已确认；`failed` 表示本次失败，`skipped` 表示历史、模板或覆盖策略跳过。
+`manual_review` 表示等待人工复核，应先在整理队列中确认执行结果，不能自动重提。
+`success` 表示本次操作被接收或完成，不能代替 `state` 判断入库；客户端不得在部分接收后原样重提整个批次。
+
 #### 站点
 
 | 方法 | 路径 | 说明 |

skills/organize-files/SKILL.md

@@ -51,6 +51,13 @@ recognition, transfer, and history tools are not available.
 - Detail: `media.detail` with `path_params.media_id` and identity/type in `query`.
 - Library check: `library.exists` with the exact identity in `query`.
 - Transfer: `transfer.file` with the manual-transfer request in `body`.
+- Actual transfer responses include `data.items` even when the batch reports
+  failure. Inspect every item's `state`: `accepted` and `retry_wait` mean that
+  background work remains; only `completed` confirms execution and settlement.
+  `manual_review` requires resolving the task in the transfer queue first;
+  `failed` and `skipped` do not mean imported. Do not resubmit an entire batch
+  that already contains accepted or completed items. Preview responses remain
+  planning data and do not contain execution states.
 - Scrape: `media.scrape` with `path_params.storage`, file item in `body`, and
   exact identity/type fields in `query`.
 

tests/test_transfer_submission.py

@@ -0,0 +1,320 @@
+"""验证实际整理提交回执与预览隔离，并由真实执行阶段决定完成状态。"""
+
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.api.endpoints import transfer
+from app.application.transfer.execution import TransferExecutionState
+from app.chain.transfer.facade import TransferChain
+from app.schemas.transfer import TransferInfo
+from tests.test_manual_transfer_history import _patch_transfer_planning
+from tests.test_transfer_sync_extra_files import make_fileitem, make_transfer_chain
+
+
+@pytest.fixture
+def submission_client(monkeypatch):
+    """挂载真实路由与响应校验，仅替换管理授权、历史端口和整理业务外部边界。"""
+    app = FastAPI()
+    app.include_router(transfer.router, prefix="/api/v1/transfer")
+    app.dependency_overrides[transfer.get_current_active_manage_user] = lambda: object()
+    app.dependency_overrides[transfer.get_transfer_history_lookup_service] = lambda: SimpleNamespace(get=lambda _id: None)
+    calls = []
+    outcomes = []
+
+    class FakeChain:
+        """按源文件返回可控回执，验证 HTTP 响应不会丢失状态和部分成功项。"""
+
+        def manual_transfer(self, **kwargs):
+            """保存请求参数并返回下一个预设结果。"""
+            calls.append(kwargs)
+            return outcomes.pop(0)
+
+    monkeypatch.setattr(transfer, "TransferChain", FakeChain)
+    with TestClient(app) as client:
+        yield client, outcomes, calls
+
+
+@pytest.mark.parametrize("state", ["accepted", "completed", "failed", "retry_wait", "skipped", "manual_review"])
+def test_submission_http_preserves_typed_state_and_failure_details(submission_client, state):
+    """实际阶段必须经过 FastAPI 序列化保留，不能降级为预览并丢失 state。"""
+    client, outcomes, calls = submission_client
+    accepted = state in {"accepted", "completed", "retry_wait"}
+    outcomes.append((accepted, {"items": [{
+        "source": "/downloads/a.mkv", "target": "/media/a.mkv", "state": state,
+        "success": accepted, "message": "目标路径不可写" if not accepted else "",
+        "overwrite_skipped": state == "skipped",
+    }]}))
+
+    response = client.post("/api/v1/transfer/manual?background=true", json={
+        "fileitem": make_fileitem("/downloads/a.mkv").model_dump(mode="json"),
+    })
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["success"] is accepted
+    item = payload["data"]["items"][0]
+    assert item["state"] == state
+    assert item["target"] == "/media/a.mkv"
+    assert item["overwrite_skipped"] is (state == "skipped")
+    if not accepted:
+        assert item["failure_stage"]
+        assert item["recovery_action"]
+    assert calls[0]["report_results"] is True
+    assert calls[0]["background"] is True
+
+
+def test_submission_batch_preserves_success_when_another_file_fails(submission_client):
+    """批次总失败仍返回完整逐项结果，并在业务总提示中保留失败文件路径。"""
+    client, outcomes, calls = submission_client
+    first, second = make_fileitem("/downloads/a.mkv"), make_fileitem("/downloads/b.mkv")
+    outcomes.extend([
+        (True, {"items": [{"source": first.path, "state": "accepted", "success": True}]}),
+        (False, "目标路径不可写"),
+    ])
+
+    response = client.post("/api/v1/transfer/manual?background=true", json={
+        "fileitems": [item.model_dump(mode="json") for item in (first, second)],
+        "target_path": "/media",
+    })
+
+    payload = response.json()
+    assert response.status_code == 200
+    assert payload["success"] is False
+    assert [item["state"] for item in payload["data"]["items"]] == ["accepted", "failed"]
+    assert [item["source"] for item in payload["data"]["items"]] == [first.path, second.path]
+    assert second.path in payload["message"]
+    assert payload["data"]["items"][1]["failure_stage"] == "destination_access"
+    assert len(calls) == 2
+
+
+def test_preview_http_keeps_its_summary_and_does_not_claim_execution(submission_client):
+    """预览继续返回原预览 schema，预览成功不得凭空生成实际接收状态。"""
+    client, outcomes, calls = submission_client
+    outcomes.append((True, {"summary": {"total": 1, "success": 1, "failed": 0}, "items": [{
+        "source": "/downloads/a.mkv", "success": True, "title": "Example",
+    }]}))
+
+    response = client.post("/api/v1/transfer/manual", json={
+        "fileitem": make_fileitem("/downloads/a.mkv").model_dump(mode="json"), "preview": True,
+    })
+
+    payload = response.json()
+    assert response.status_code == 200
+    assert payload["data"]["summary"]["total"] == 1
+    assert payload["data"]["items"][0]["title"] == "Example"
+    assert "state" not in payload["data"]["items"][0]
+    assert calls[0]["report_results"] is False
+
+
+@pytest.fixture
+def submission_chain(monkeypatch):
+    """保留真实手动工作流，仅隔离文件、下载器、媒体识别和执行副作用。"""
+    chain = make_transfer_chain()
+    fileitem = make_fileitem("/downloads/Test.Show.S01E01.mkv")
+    _patch_transfer_planning(monkeypatch, chain, fileitem, None, [], [])
+    monkeypatch.setattr(chain, "_TransferChain__claim_task_for_execution", lambda task: None)
+    monkeypatch.setattr(chain, "_TransferChain__start_job_execution", lambda task: None)
+    monkeypatch.setattr(chain, "_TransferChain__release_task_claim", lambda task, **kwargs: None)
+    return chain, fileitem
+
+
+def test_background_submission_reports_queue_acceptance_without_execution(submission_chain, monkeypatch):
+    """后台成功仅确认真正入队，提交请求本身不会执行文件也不能报告 completed。"""
+    chain, fileitem = submission_chain
+    queued = []
+    monkeypatch.setattr(chain, "put_to_queue", lambda task: queued.append(task) or True)
+    success, result = chain.manual_transfer(fileitem=fileitem, background=True, report_results=True)
+    assert success is True
+    assert len(queued) == 1
+    assert result["items"][0]["state"] == "accepted"
+    assert result["items"][0]["source"] == fileitem.path
+
+
+@pytest.mark.parametrize("queued, expected", [(False, "failed"), (True, "accepted")])
+def test_background_duplicate_is_not_reported_as_accepted(submission_chain, monkeypatch, queued, expected):
+    """只有队列入口确认新接收才返回 accepted，重复准入必须明确失败。"""
+    chain, fileitem = submission_chain
+    monkeypatch.setattr(chain, "put_to_queue", lambda task: queued)
+    success, result = chain.manual_transfer(fileitem=fileitem, background=True, report_results=True)
+    assert success is queued
+    assert result["items"][0]["state"] == expected
+
+
+@pytest.mark.parametrize("settled, overwrite, expected", [
+    (True, False, "completed"), (False, False, "retry_wait"), (True, True, "skipped"),
+])
+def test_sync_submission_waits_for_atomic_settlement(submission_chain, monkeypatch, settled, overwrite, expected):
+    """执行成功但原子结算未确认时必须等待恢复，覆盖跳过也不能被误报已入库。"""
+    chain, fileitem = submission_chain
+    target = make_fileitem("/media/Test.Show.S01E01.mkv")
+
+    def settle(task, info):
+        """模拟 writer 的明确结算回执，保留工作流中真实完成判定。"""
+        if settled:
+            task.mark_terminal_settled()
+        return info.success, info.message or ""
+
+    def execute(task, callback):
+        """模拟一次文件操作并进入真实工作流回调包装。"""
+        task.bind_plan_checkpoint(SimpleNamespace())
+        task.bind_admission_task_id("test-submission")
+        return callback(task, TransferInfo(
+            success=not overwrite, target_item=target,
+            message="目标已存在，跳过覆盖" if overwrite else "", overwrite_skipped=overwrite,
+        ))
+
+    monkeypatch.setattr(chain, "_TransferChain__default_callback", settle)
+    monkeypatch.setattr(chain, "_TransferChain__handle_transfer", execute)
+    chain.transfer_execution_repository = SimpleNamespace(
+        get_snapshot=lambda **kwargs: SimpleNamespace(state=TransferExecutionState.RETRY_WAIT),
+    )
+    _success, result = chain.manual_transfer(fileitem=fileitem, background=False, report_results=True)
+    assert result["items"][0]["state"] == expected
+    assert result["items"][0]["target"] == target.path
+    assert result["items"][0]["overwrite_skipped"] is overwrite
+
+
+@pytest.mark.parametrize("accepted, expected", [(True, "retry_wait"), (False, "failed")])
+def test_existing_durable_retry_keeps_its_actual_state(submission_chain, monkeypatch, accepted, expected):
+    """历史失败的原计划重试由现有调度器接管，不标记为同步完成。"""
+    chain, fileitem = submission_chain
+    history = SimpleNamespace(status=False, transfer_task_id="task-a")
+    monkeypatch.setattr(chain, "_get_manual_transfer_history", lambda **kwargs: history)
+    monkeypatch.setattr(chain, "_request_durable_transfer_retry", lambda *args, **kwargs: (accepted, "重试申请"))
+    success, result = chain.manual_transfer(fileitem=fileitem, background=False, report_results=True)
+    assert success is accepted
+    assert result["items"][0]["state"] == expected
+
+
+def test_legacy_manual_caller_keeps_string_result(submission_chain):
+    """未启用回执的插件和既有调用仍收到原来的布尔与字符串二元组。"""
+    chain, fileitem = submission_chain
+    success, message = chain.manual_transfer(fileitem=fileitem, background=False)
+    assert success is True
+    assert message == ""
+
+
+@pytest.mark.parametrize("background", [True, False])
+def test_cancellation_keeps_processed_and_unprocessed_file_receipts(submission_chain, monkeypatch, background):
+    """中途取消保留已入队或完成项，同时为剩余文件返回失败，禁止整批重提。"""
+    chain, first = submission_chain
+    second = make_fileitem("/downloads/Test.Show.S01E02.mkv")
+    stop = SimpleNamespace(is_system_stopped=False)
+    registered = []
+    monkeypatch.setattr("app.chain.transfer.workflow.runtime_stop_state", stop)
+    monkeypatch.setattr(chain, "_TransferChain__get_trans_fileitems", lambda *args, **kwargs: [(first, False), (second, False)])
+
+    def register(task):
+        """使用真实作业视图记录等待任务，以验证取消后可重新准入。"""
+        registered.append(task)
+        return TransferChain._TransferChain__put_to_jobview(chain, task)
+
+    def queue(task):
+        """第一个文件真实被接收后模拟用户取消整个批次。"""
+        assert task.fileitem.path == first.path
+        stop.is_system_stopped = True
+        return True
+
+    def execute(task, callback):
+        """第一个文件已结算后模拟取消，不执行剩余文件。"""
+        assert callback
+        assert task.fileitem.path == first.path
+        task.mark_terminal_settled()
+        stop.is_system_stopped = True
+        return True, ""
+
+    monkeypatch.setattr(chain, "put_to_queue", queue)
+    monkeypatch.setattr(chain, "_TransferChain__handle_transfer", execute)
+    monkeypatch.setattr(chain, "_TransferChain__put_to_jobview", register)
+    success, result = chain.manual_transfer(fileitem=first, background=background, report_results=True)
+    assert success is False
+    assert [(item["source"], item["state"]) for item in result["items"]] == [
+        (first.path, "accepted" if background else "completed"), (second.path, "failed"),
+    ]
+    assert "取消" in result["items"][1]["message"]
+    if not background:
+        assert TransferChain._TransferChain__put_to_jobview(chain, registered[1]) is True
+
+
+def test_post_settlement_failure_preserves_completed_receipt(submission_chain, monkeypatch):
+    """入库原子提交后通知等后续副作用失败，回执仍确认已入库，避免重复整理。"""
+    chain, fileitem = submission_chain
+
+    def settle(task, _info):
+        """原子提交成功后模拟通知异常。"""
+        task.mark_terminal_settled()
+        raise RuntimeError("通知失败")
+
+    def execute(task, callback):
+        """把成功文件操作交给真实回调包装，保留提交前后的时间顺序。"""
+        return callback(task, TransferInfo(success=True, target_item=make_fileitem("/media/done.mkv")))
+
+    monkeypatch.setattr(chain, "_TransferChain__default_callback", settle)
+    monkeypatch.setattr(chain, "_TransferChain__handle_transfer", execute)
+    monkeypatch.setattr(chain, "_TransferChain__fail_transfer_task", lambda *args: None)
+    success, result = chain.manual_transfer(fileitem=fileitem, background=False, report_results=True)
+    assert success is False
+    assert result["items"][0]["state"] == "completed"
+    assert result["items"][0]["success"] is True
+    assert result["items"][0]["target"] == "/media/done.mkv"
+    assert "文件已完成整理" in result["items"][0]["message"]
+    assert "无需重复整理" in result["items"][0]["message"]
+    assert "稍后重试" not in result["items"][0]["message"]
+
+
+def test_cancellation_during_build_releases_waiting_job(submission_chain, monkeypatch):
+    """同步任务尚未开始执行时取消，也要移除预登记占位，随后允许重新准入。"""
+    chain, first = submission_chain
+    second = make_fileitem("/downloads/Test.Show.S01E02.mkv")
+    stop = SimpleNamespace(is_system_stopped=False)
+    registered = []
+    monkeypatch.setattr("app.chain.transfer.workflow.runtime_stop_state", stop)
+    monkeypatch.setattr(chain, "_TransferChain__get_trans_fileitems", lambda *args, **kwargs: [(first, False), (second, False)])
+
+    def register(task):
+        """第一个等待项登记后取消，模拟候选构建阶段被用户中断。"""
+        registered.append(task)
+        queued = TransferChain._TransferChain__put_to_jobview(chain, task)
+        stop.is_system_stopped = True
+        return queued
+
+    monkeypatch.setattr(chain, "_TransferChain__put_to_jobview", register)
+    success, result = chain.manual_transfer(fileitem=first, background=False, report_results=True)
+    assert success is False
+    assert [item["state"] for item in result["items"]] == ["failed", "failed"]
+    assert len(registered) == 1
+    assert TransferChain._TransferChain__put_to_jobview(chain, registered[0]) is True
+
+
+@pytest.mark.parametrize("persisted, expected", [
+    (TransferExecutionState.NOT_STARTED, "retry_wait"),
+    (TransferExecutionState.RETRY_WAIT, "retry_wait"),
+    (TransferExecutionState.MANUAL_REVIEW, "manual_review"),
+    (None, "failed"),
+])
+@pytest.mark.parametrize("planned", [True, False])
+def test_execution_failure_uses_remaining_durable_state(submission_chain, monkeypatch, persisted, expected, planned):
+    """检查点前失败、延迟重试、人工复核与已清理损坏任务分别保留真实恢复语义。"""
+    chain, fileitem = submission_chain
+
+    def execute(task, callback):
+        """模拟已准入任务在完成检查点前离开执行流程。"""
+        assert callback
+        task.bind_admission_task_id("test-submission")
+        if planned:
+            task.bind_plan_checkpoint(SimpleNamespace())
+        raise RuntimeError("执行暂时失败")
+
+    monkeypatch.setattr(chain, "_TransferChain__handle_transfer", execute)
+    monkeypatch.setattr(chain, "_TransferChain__fail_transfer_task", lambda *args: None)
+    chain.transfer_execution_repository = SimpleNamespace(
+        get_snapshot=lambda **kwargs: SimpleNamespace(state=persisted) if persisted else None,
+    )
+    success, result = chain.manual_transfer(fileitem=fileitem, background=False, report_results=True)
+    assert success is False
+    assert result["items"][0]["state"] == expected
+    if expected == "manual_review":
+        assert "人工复核" in result["items"][0]["recovery_action"]