feat(transfer): implement recovery commands for discarding corrupt tasks

app/api/endpoints/history.py

@@ -44,6 +44,7 @@ from app.application.transfer.execution import (
     TransferExecutionRepository,
     TransferRetryRequestResult,
 )
+from app.application.transfer.recovery import TransferRecoveryCommand
 from app.runtime.log import logger
 from app.runtime.loop import main_loop_registry
 from app.runtime.progress import AsyncProgressHelper
@@ -54,6 +55,7 @@ from app.schemas.history import BatchTransferHistoryRedoRequest as _SchemaBatchT
 from app.schemas.history import DownloadHistory as _SchemaDownloadHistory
 from app.schemas.history import TransferHistory as _SchemaTransferHistory
 from app.schemas.history import TransferHistoryDeleteResult as _SchemaTransferHistoryDeleteResult
+from app.schemas.history import TransferHistoryDiscardResult as _SchemaTransferHistoryDiscardResult
 from app.schemas.history import TransferHistoryPage as _SchemaTransferHistoryPage
 from app.schemas.response import Response as _SchemaResponse
 from app.schemas.token import TokenPayload as _SchemaTokenPayload
@@ -457,29 +459,42 @@ def delete_transfer_history(
     )
 
 
-@router.post(
-    "/transfer/{history_id}/discard-corrupt",
-    summary="放弃损坏的整理任务",
-    response_model=_SchemaResponse[dict],
-)
-async def discard_corrupt_transfer_history(
+async def _get_discard_transfer_history(
     history_id: int,
     query: HistoryQueryService = Depends(get_history_query_service),
+) -> Optional[_SchemaTransferHistory]:
+    """异步加载历史 DTO，随后由 FastAPI 工作线程执行同步原子清理。"""
+    return await query.get_transfer(history_id)
+
+
+# FastAPI 装饰器在当前 mypy 配置下视为无类型，响应仍由具体 Pydantic 模型约束。
+@router.post(  # type: ignore[misc]
+    "/transfer/{history_id}/discard-corrupt",
+    summary="放弃损坏的整理任务",
+    response_model=_SchemaResponse[_SchemaTransferHistoryDiscardResult],
+)
+def discard_corrupt_transfer_history(
+    history_id: int,
+    history: Optional[_SchemaTransferHistory] = Depends(_get_discard_transfer_history),
     execution_repository: TransferExecutionRepository = Depends(get_transfer_execution_repository),
     _: object = Depends(get_current_active_manage_user),
 ) -> Any:
     """清理无活动租约的损坏 durable 任务，保留历史供重新生成计划。"""
-    history = await query.get_transfer(history_id)
     if not history:
         return _SchemaResponse(success=False, message="整理记录不存在")
     if not history.transfer_task_id:
-        return _SchemaResponse(success=True, message="整理任务已清理", data={"history_id": history_id})
-    result = await asyncio.to_thread(
-        TransferExecutionCommand(execution_repository).discard_corrupt_by_history,
+        return _SchemaResponse(
+            success=True, message="整理任务已清理",
+            data=_SchemaTransferHistoryDiscardResult(history_id=history_id),
+        )
+    result = TransferRecoveryCommand(execution_repository).discard_corrupt_by_history(
         task_id=history.transfer_task_id,
         history_id=history_id,
     )
-    return _SchemaResponse(success=result.discarded, message=result.message, data={"history_id": history_id})
+    return _SchemaResponse(
+        success=result.discarded, message=result.message,
+        data=_SchemaTransferHistoryDiscardResult(history_id=history_id),
+    )
 
 
 @router.post(

app/application/historymutation.py

@@ -7,9 +7,9 @@ from pathlib import Path
 from typing import Any, Callable, Optional, Protocol
 
 from app.application.transfer.execution import (
-    TransferExecutionCommand,
     TransferExecutionRepository,
 )
+from app.application.transfer.recovery import TransferRecoveryCommand
 from app.schemas.common import JsonData
 from app.schemas.history import TransferHistoryDeleteResult, TransferHistoryDeleteStep
 
@@ -185,7 +185,7 @@ class TransferHistoryMutationCommand:
                     history="retained",
                     message="持久整理失败记录缺少结算版本，请刷新后重试",
                 )
-            discard = TransferExecutionCommand(
+            discard = TransferRecoveryCommand(
                 self._transfer_execution_repository
             ).discard_failed(
                 task_id=history.transfer_task_id,

app/application/plugin/data.py

@@ -1,6 +1,7 @@
-from __future__ import annotations
 """插件持久化数据查询、投影与写用例。"""
 
+from __future__ import annotations
+
 import json
 from collections.abc import Callable
 from typing import Any, Optional, Protocol

app/application/subscription/execution.py

@@ -1,6 +1,7 @@
-from __future__ import annotations
 """订阅执行准入、搜索上下文、批次任务与持久队列端口。"""
 
+from __future__ import annotations
+
 import threading
 import time
 from dataclasses import dataclass

app/application/transfer/execution.py

@@ -849,49 +849,6 @@ class TransferExecutionCommand:
             requested_by=requested_by,
         )
 
-    def discard_corrupt_task(
-            self,
-            *,
-            task_id: str,
-            lease_token: str,
-            error: str,
-    ) -> bool:
-        """以当前租约清理损坏任务，避免恢复线程再次回放旧步骤。"""
-        if not task_id or not lease_token or not error:
-            raise ValueError("损坏任务收口缺少任务、租约或错误原因")
-        return self._repository.discard_corrupt_task(
-            task_id=task_id, lease_token=lease_token, error=error
-        )
-
-    def discard_corrupt_by_history(
-            self,
-            *,
-            task_id: str,
-            history_id: int,
-    ) -> TransferFailureDiscardResult:
-        """放弃无法重试的损坏任务，保留历史记录供重新生成计划。"""
-        if not task_id or history_id <= 0:
-            raise ValueError("放弃损坏任务缺少任务或历史")
-        return self._repository.discard_corrupt_by_history(
-            task_id=task_id, history_id=history_id
-        )
-
-    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,
             *,

app/application/transfer/recovery.py

@@ -0,0 +1,73 @@
+"""失败与损坏整理任务的证据清理和历史解绑用例。"""
+
+from typing import Optional
+
+from app.application.transfer.execution import (
+    TransferExecutionConflictError,
+    TransferExecutionRepository,
+    TransferFailureDiscardResult,
+)
+
+
+class TransferRecoveryCommand:
+    """把损坏任务恢复交给持久层原子清理，保留历史供重新规划。"""
+
+    def __init__(self, repository: TransferExecutionRepository) -> None:
+        """保存拥有租约校验与事务的执行仓储。"""
+        self._repository = repository
+
+    def discard_corrupt_task(
+            self,
+            *,
+            task_id: str,
+            lease_token: str,
+            error: str,
+    ) -> bool:
+        """以当前租约清理损坏任务，避免恢复线程再次回放旧步骤。"""
+        if not task_id or not lease_token or not error:
+            raise ValueError("损坏任务收口缺少任务、租约或错误原因")
+        return self._repository.discard_corrupt_task(
+            task_id=task_id, lease_token=lease_token, error=error
+        )
+
+    def discard_corrupt_by_history(
+            self,
+            *,
+            task_id: str,
+            history_id: int,
+    ) -> TransferFailureDiscardResult:
+        """放弃无法重试的损坏任务，保留历史记录供重新生成计划。"""
+        if not task_id or history_id <= 0:
+            raise ValueError("放弃损坏任务缺少任务或历史")
+        return self._repository.discard_corrupt_by_history(
+            task_id=task_id, history_id=history_id
+        )
+
+    def discard_conflict(
+        self, *, task_id: Optional[str], lease_token: Optional[str], error: object,
+    ) -> bool:
+        """只清理有租约且无法恢复的规划冲突，普通执行错误仍保留重试证据。"""
+        if not isinstance(error, TransferExecutionConflictError) or not task_id or not lease_token:
+            return False
+        message = str(error)
+        if not any(marker in message for marker in (
+            "记录已失效", "记录不完整", "版本不一致", "检查点", "恢复状态不完整",
+        )):
+            return False
+        return self.discard_corrupt_task(task_id=task_id, lease_token=lease_token, error=message)
+
+    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,
+        )

app/chain/download/subtitle.py

@@ -1,6 +1,7 @@
-from __future__ import annotations
 """字幕获取、解压和存储 owner。"""
 
+from __future__ import annotations
+
 import re
 import shutil
 import time

app/chain/transfer/records.py

@@ -17,9 +17,9 @@ from app.application.history import (
     resolve_history,
 )
 from app.application.transfer.execution import (
-    TransferExecutionCommand,
     TransferExecutionRepository,
 )
+from app.application.transfer.recovery import TransferRecoveryCommand
 from app.chain._contracts import TransferMixinHost
 from app.chain.storage import StorageChain
 from app.chain.subscribe.facade import SubscribeChain
@@ -528,7 +528,7 @@ class ManualHistoryMixin(_TransferOwnerBase):
             )
             if not settlement_revision:
                 return False, "持久整理失败记录缺少结算版本，请刷新后重试"
-            discard = TransferExecutionCommand(
+            discard = TransferRecoveryCommand(
                 self.transfer_execution_repository
             ).discard_failed(
                 task_id=task_id,

app/chain/transfer/settlement.py

@@ -24,8 +24,10 @@ from app.application.outbox import (
 )
 from app.application.transfer.execution import (
     TransferExecutionConflictError,
+    TransferExecutionRepository,
     TransferSettlementResult,
 )
+from app.application.transfer.recovery import TransferRecoveryCommand
 from app.application.transfer.workflow import (
     TransferFailureNotification,
     TransferLeaseLostError,
@@ -51,6 +53,19 @@ from app.schemas.types import (
 )
 
 
+def _discard_corrupt_transfer_task(
+    repository: TransferExecutionRepository, task: TransferTask, error: object,
+) -> None:
+    """清理失败时继续结束内存作业，持久层保留的证据仍可用于后续恢复。"""
+    if not task.preview:
+        try:
+            TransferRecoveryCommand(repository).discard_conflict(
+                task_id=task.admission_task_id, lease_token=task.lease_token, error=error,
+            )
+        except Exception as cleanup_error:
+            logger.error(f"清理损坏整理任务 durable 证据失败：{task.admission_task_id} - {cleanup_error}")
+
+
 class TransferSettlementOwner(_TransferOwnerBase):
     """唯一持有整理终态结算、历史事件和失败通知。"""
 
@@ -695,29 +710,7 @@ class TransferSettlementOwner(_TransferOwnerBase):
 
     def _TransferChain__fail_transfer_task(self, task: TransferTask, error: object = "整理任务处理失败"):
         """清理作业视图，并在执行冲突时原子删除 durable 恢复证据。"""
-        error_text = str(error)
-        corrupt_plan = any(
-            marker in error_text
-            for marker in ("记录已失效", "记录不完整", "版本不一致", "检查点", "恢复状态不完整")
-        )
-        if (
-                isinstance(error, TransferExecutionConflictError)
-                and corrupt_plan
-                and not task.preview
-                and task.admission_task_id
-                and task.lease_token
-        ):
-            try:
-                self._transfer_executions.discard_corrupt_task(
-                    task_id=task.admission_task_id,
-                    lease_token=task.lease_token,
-                    error=str(error),
-                )
-            except Exception as cleanup_error:
-                logger.error(
-                    "清理损坏整理任务 durable 证据失败：%s - %s",
-                    task.admission_task_id, cleanup_error,
-                )
+        _discard_corrupt_transfer_task(self.transfer_execution_repository, task, error)
         self.jobview.fail_unfinished_task(task)
         self.jobview.try_remove_job(task)
         self._finish_scrape_batch_task(task)

app/locales/en-US.json

@@ -103,6 +103,7 @@
     }
   },
   "messages": {
+    "整理任务已清理": "The organization task has been cleared",
     "暂无到期订阅，无需搜索": "No subscriptions are due for search",
     "调用工具失败": "Tool call failed",
     "无效的媒体来源": "Invalid media source",

app/locales/zh-CN.json

@@ -103,6 +103,7 @@
     }
   },
   "messages": {
+    "整理任务已清理": "整理任务已清理",
     "当前管理用户缺少可审计身份": "当前管理用户缺少可审计身份",
     "人工复核任务不存在": "人工复核任务不存在",
     "媒体来源和媒体 ID 必须同时提供": "媒体来源和媒体 ID 必须同时提供",

app/locales/zh-TW.json

@@ -103,6 +103,7 @@
     }
   },
   "messages": {
+    "整理任务已清理": "整理任務已清理",
     "暂无到期订阅，无需搜索": "暫無到期訂閱，無需搜尋",
     "调用工具失败": "調用工具失敗",
     "媒体来源和媒体 ID 必须同时提供": "媒體來源和媒體 ID 必須同時提供",

app/schemas/exports.py

@@ -500,6 +500,7 @@ SCHEMA_EXPORTS = {
     'TransferHistory': ('app.schemas.history', 'TransferHistory'),
     'TransferHistoryDeleteResult': ('app.schemas.history', 'TransferHistoryDeleteResult'),
     'TransferHistoryDeleteStep': ('app.schemas.history', 'TransferHistoryDeleteStep'),
+    'TransferHistoryDiscardResult': ('app.schemas.history', 'TransferHistoryDiscardResult'),
     'TransferHistoryPage': ('app.schemas.history', 'TransferHistoryPage'),
     'TransferInfo': ('app.schemas.transfer', 'TransferInfo'),
     'TransferInterceptEventData': ('app.schemas.event', 'TransferInterceptEventData'),

app/schemas/history.py

@@ -167,6 +167,12 @@ class BatchTransferHistoryRedoRequest(BaseModel):
     history_ids: list[int] = Field(default_factory=list)
 
 
+class TransferHistoryDiscardResult(BaseModel):  # type: ignore[misc]
+    """损坏整理任务清理结果，标识保留供后续操作的历史记录。"""
+
+    history_id: int = Field(description="保留的整理历史记录 ID")
+
+
 class TransferHistoryPage(BaseModel):
     """整理历史分页数据。"""
 

docs/architecture-overview.md

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

docs/architecture/agent-api-surface-audit.json

@@ -6,7 +6,7 @@
     "provider-skill": 12,
     "stream_or_binary": 10,
     "transport_or_identity": 66,
-    "ui_presentation": 20
+    "ui_presentation": 21
   },
   "dynamic_gateway_routes": [
     {
@@ -21,7 +21,7 @@
   "gateway_http_route_count": 203,
   "gateway_operation_count": 205,
   "matched_gateway_http_route_count": 202,
-  "openapi_operation_count": 393,
+  "openapi_operation_count": 394,
   "operations": [
     {
       "disposition": "consolidated",
@@ -837,6 +837,18 @@
         "history"
       ]
     },
+    {
+      "disposition": "ui_presentation",
+      "method": "POST",
+      "operation_ids": [],
+      "owner": "host-ui",
+      "path": "/api/v1/history/transfer/{history_id}/discard-corrupt",
+      "reason": "Discarding corrupt transfer state is owned by the authenticated management recovery workflow; it is not a stable Agent gateway operation.",
+      "summary": "放弃损坏的整理任务",
+      "tags": [
+        "history"
+      ]
+    },
     {
       "disposition": "transport_or_identity",
       "method": "POST",

docs/architecture/agent-api-surface-audit.md

@@ -5,7 +5,7 @@
 
 ## Result
 
-- OpenAPI HTTP operations: **393**
+- OpenAPI HTTP operations: **394**
 - Stable `moviepilot_api` operations: **205**
 - Exact HTTP routes used by the gateway: **203**
 - OpenAPI routes matched directly by the gateway: **202**
@@ -23,7 +23,7 @@
 | `provider-skill` | 12 | Low-level downloader or media-server capability owned by a provider Skill. |
 | `stream_or_binary` | 10 | Streaming or binary response owned by a direct client transport. |
 | `transport_or_identity` | 66 | Authentication, protocol, callback, account, or conversation transport boundary. |
-| `ui_presentation` | 20 | Frontend or plugin-rendered presentation contract. |
+| `ui_presentation` | 21 | Frontend or plugin-rendered presentation contract. |
 
 ## Bounded Dynamic Routes
 
@@ -99,6 +99,7 @@
 | `POST` | `/api/v1/history/transfer/ai-redo` | history | `gateway` | transfer.history.redo_batch | 智能助手批量重新整理 |
 | `DELETE` | `/api/v1/history/transfer/all` | history | `gateway` | transfer.history.clear | 清空旧整理记录 |
 | `POST` | `/api/v1/history/transfer/{history_id}/ai-redo` | history | `gateway` | transfer.history.redo | 智能助手重新整理 |
+| `POST` | `/api/v1/history/transfer/{history_id}/discard-corrupt` | history | `ui_presentation` | host-ui | 放弃损坏的整理任务 |
 | `POST` | `/api/v1/llm/manage` | llm | `transport_or_identity` | host-runtime | LLM提供商统一管理 |
 | `GET` | `/api/v1/llm/provider-auth/callback/{provider_id}` | llm | `transport_or_identity` | host-runtime | LLM提供商OAuth回调 |
 | `POST` | `/api/v1/login/access-token` | login | `transport_or_identity` | host-runtime | 获取token |

docs/architecture/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 983 / 8,340 | `dependency-baseline.json` 当前快照；分类离线词表、下载资源分类与订阅搜索运行时任务新增模块及其受控依赖 |
+| 宿主 Python 模块 / 内部依赖边 | 985 / 8,356 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索与整理任务恢复模块的受控依赖 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

docs/mcp-api.md

@@ -19,6 +19,8 @@ MCP 使用系统配置中的 `API_TOKEN` 作为认证密钥，文档中的 API K
 - 不要在缺少 HTTPS、访问控制和网络隔离的情况下，将 MCP、OpenAI 或 Anthropic 兼容接口直接暴露到公网。
 - MCP 隐藏工具列表只用于减少默认暴露面，不是 per-user 权限系统。
 
+`POST /api/v1/history/transfer/{history_id}/discard-corrupt` 属于需要管理权限的整理恢复 REST 接口，不向 Agent gateway 暴露。成功响应的 `data.history_id` 为保留的整理历史 ID；任务已清理时同样返回该结构，历史不存在时返回业务失败。
+
 ## 2. 标准 MCP 协议 (JSON-RPC 2.0)
 
 ### 端点
@@ -52,25 +54,25 @@ MCP 当前不会主动发送工具列表变更通知（`listChanged=false`）。
 
 `app/agent/policy/resources/api_mcp_schema.json` 是 `moviepilot_api` 的生成制品，不是设置项或 API 参数的手工事实源。`scripts/generate_agent_api_mcp_schema.py` 从当前 FastAPI OpenAPI、固定 operation 路由和 Agent 专用英文参数说明生成该文件；运行时直接读取它响应外部 MCP `tools/list`，测试会校验生成结果没有漂移。修改 API、请求模型或 operation 后应重新生成并提交该文件，不应直接编辑 JSON。
 
-当前完整 FastAPI OpenAPI 包含 375 个 HTTP 操作，其中 203 个稳定业务操作进入
-`moviepilot_api`，使用 201 个固定路由模板：200 条 OpenAPI 路由直接匹配，另有 1 条只允许
+当前完整 FastAPI OpenAPI 包含 394 个 HTTP 操作，其中 205 个稳定业务操作进入
+`moviepilot_api`，使用 203 个固定路由模板：202 条 OpenAPI 路由直接匹配，另有 1 条只允许
 `tmdb`、`douban`、`bangumi`、`anilist` 四个来源的受限人物作品动态路由。每个 operation
 均同时具备固定 method/path、角色权限、副作用等级、确认与恢复策略、结果敏感性、英文用途说明，
 以及可直接提交的 path/query/body JSON Schema；Skill front matter、正文 operation 章节、运行时
-注册表和 MCP `tools/list` 的 203 个 `oneOf` 分支必须完全一致。
+注册表和 MCP `tools/list` 的 205 个 `oneOf` 分支必须完全一致。
 
-数量不相等是明确的安全与语义边界，而不是漏生成。当前 375 条路由均被审计并锁定为以下一种
+数量不相等是明确的安全与语义边界，而不是漏生成。当前 394 条路由均被审计并锁定为以下一种
 归属，审计生成器不再提供“未归类”兜底：
 
 | 归属 | 数量 | Agent 使用方式 |
 | :--- | ---: | :--- |
-| `gateway` | 200 | 通过 `moviepilot_api` 的稳定 operation 和精确参数合同调用 |
+| `gateway` | 202 | 通过 `moviepilot_api` 的稳定 operation 和精确参数合同调用 |
 | `consolidated` | 72 | 通过同领域聚合 operation 调用，不复制数据源或前端专用路由 |
-| `provider-skill` | 11 | 通过下载器或媒体服务器 Skill 调用第三方 provider API |
+| `provider-skill` | 12 | 通过下载器或媒体服务器 Skill 调用第三方 provider API |
 | `alternate-auth-duplicate` | 11 | 使用对应 bearer-authenticated gateway operation，不暴露 API_TOKEN 兼容副本 |
 | `transport_or_identity` | 66 | 由登录、令牌、MCP、会话、回调、健康检查等宿主传输/身份边界拥有 |
 | `stream_or_binary` | 10 | 由直接客户端处理流式日志、消息、文件、图片等非结构化响应 |
-| `ui_presentation` | 5 | 由前端或插件渲染面拥有，不作为业务 Agent operation |
+| `ui_presentation` | 21 | 由前端或插件渲染面拥有，不作为业务 Agent operation |
 
 逐路由归属见 `docs/architecture/agent-api-surface-audit.md`，并由
 `tests/test_agent_api_surface_audit.py` 对当前 OpenAPI、固定注册表、MCP schema、英文 Skill

docs/rules/05-architecture.md

@@ -70,7 +70,7 @@ to make the directory tree look symmetrical.
 | `app/application/agent.py` | Agent orchestration facade and typed `AgentDataContext`; startup injects one explicit data context into the manager, memory, tool and scheduler owners without a process-wide persistence locator |
 | `app/application/network.py` | System network-test target catalog, immutable public/private projections, URL and redirect admission, response validation and the injected transport Port; startup owns concrete HTTP Adapter assembly |
 | `app/application/outbox.py` | Durable intent, transaction-only stager, short-transaction dispatch store, claim fencing and structured post-commit result contracts |
-| `app/application/transfer/` | Durable transfer use cases: `workflow.py` owns admission/planning/queue behavior; `execution.py` owns stable operation identity, step/checkpoint state, retry/manual-review commands and terminal-settlement DTOs |
+| `app/application/transfer/` | Durable transfer use cases: `workflow.py` owns admission/planning/queue behavior; `execution.py` owns stable operation identity, step/checkpoint state, retry/manual-review commands and terminal-settlement DTOs; `recovery.py` owns failed/corrupt task cleanup and history detachment through the execution repository |
 | `app/application/plugin/` | Plugin market catalog, installation command, installed-plugin identity contract and startup migration, runtime port, folder operations and dynamic-route use cases; filenames remain single words (`catalog.py`, `identity.py`, `migration.py`, `install.py`, `runtime.py`, `folders.py`, `routes.py`) |
 | `app/application/server/` | MoviePilot Server reporting and sharing use cases; local data readers and transport callbacks are injected by startup |
 | `app/application/site/` | Configured site catalog, authentication level and index-resource capability; the generated extension and its data bundle stay together here |

scripts/generate_agent_api_surface_audit.py

@@ -168,6 +168,13 @@ def _classify(
             "Plugin-rendered page, dashboard, or navigation metadata owned by the frontend presentation contract rather than an Agent business action.",
             [],
         )
+    if path == "/api/v1/history/transfer/{history_id}/discard-corrupt":
+        return (
+            "ui_presentation",
+            "host-ui",
+            "Discarding corrupt transfer state is owned by the authenticated management recovery workflow; it is not a stable Agent gateway operation.",
+            [],
+        )
     if path.startswith(SUBSCRIPTION_EXECUTION_UI_PREFIX):
         return (
             "ui_presentation",

skills/moviepilot-api/SKILL.md

@@ -101,6 +101,10 @@ The maintained route-by-route inventory is
 `docs/architecture/agent-api-surface-audit.md`. Its generated drift test fails
 when OpenAPI changes without an explicit ownership decision.
 
+The management recovery route `POST /api/v1/history/transfer/{history_id}/discard-corrupt`
+is reserved for direct authenticated management clients and is not a callable
+Agent operation. It clears corrupt task state while retaining the history record.
+
 ## Calling Contract
 
 Call the gateway with this shape:

tests/fixtures/architecture/dependency-baseline.json

@@ -1074,8 +1074,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8340,
-  "edge_sha256": "b3092e1b2d83dd6b9356e1107e034c1a5b65f2246a5677f7457687a37aeba4cb",
+  "edge_count": 8356,
+  "edge_sha256": "62b05583fcf296d4f33f9c5ba77eeb8f5a8a200a3809e74ad4df28726587ece4",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -2261,6 +2261,7 @@
     "app.api.endpoints.history -> app.application.history",
     "app.api.endpoints.history -> app.application.transfer",
     "app.api.endpoints.history -> app.application.transfer.execution",
+    "app.api.endpoints.history -> app.application.transfer.recovery",
     "app.api.endpoints.history -> app.runtime",
     "app.api.endpoints.history -> app.runtime.errors",
     "app.api.endpoints.history -> app.runtime.log",
@@ -3255,6 +3256,7 @@
     "app.application.historymutation -> app.application",
     "app.application.historymutation -> app.application.transfer",
     "app.application.historymutation -> app.application.transfer.execution",
+    "app.application.historymutation -> app.application.transfer.recovery",
     "app.application.historymutation -> app.schemas",
     "app.application.historymutation -> app.schemas.common",
     "app.application.historymutation -> app.schemas.history",
@@ -3808,6 +3810,9 @@
     "app.application.transfer.history -> app.domain.meta.metamusic",
     "app.application.transfer.history -> app.schemas",
     "app.application.transfer.history -> app.schemas.file",
+    "app.application.transfer.recovery -> app.application",
+    "app.application.transfer.recovery -> app.application.transfer",
+    "app.application.transfer.recovery -> app.application.transfer.execution",
     "app.application.transfer.workflow -> app.application",
     "app.application.transfer.workflow -> app.application.history",
     "app.application.transfer.workflow -> app.application.transfer",
@@ -5177,6 +5182,7 @@
     "app.chain.transfer.records -> app.application.history",
     "app.chain.transfer.records -> app.application.transfer",
     "app.chain.transfer.records -> app.application.transfer.execution",
+    "app.chain.transfer.records -> app.application.transfer.recovery",
     "app.chain.transfer.records -> app.chain",
     "app.chain.transfer.records -> app.chain._contracts",
     "app.chain.transfer.records -> app.chain.storage",
@@ -5252,6 +5258,7 @@
     "app.chain.transfer.settlement -> app.application.outbox",
     "app.chain.transfer.settlement -> app.application.transfer",
     "app.chain.transfer.settlement -> app.application.transfer.execution",
+    "app.chain.transfer.settlement -> app.application.transfer.recovery",
     "app.chain.transfer.settlement -> app.application.transfer.workflow",
     "app.chain.transfer.settlement -> app.chain",
     "app.chain.transfer.settlement -> app.chain.storage",
@@ -9427,7 +9434,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 983,
+  "module_count": 985,
   "modules": [
     "app",
     "app.adapters",
@@ -9761,6 +9768,7 @@
     "app.application.transfer.checkpoint",
     "app.application.transfer.execution",
     "app.application.transfer.history",
+    "app.application.transfer.recovery",
     "app.application.transfer.workflow",
     "app.application.workflow",
     "app.chain",

tests/fixtures/architecture/startup-performance-baseline.json

@@ -6,7 +6,7 @@
   "repeat": 3,
   "targets": {
     "app.startup.lifecycle": {
-      "loaded_app_module_count": 540,
+      "loaded_app_module_count": 541,
       "max_ms": 1293.338,
       "median_ms": 1156.239,
       "min_ms": 1102.806,
@@ -17,7 +17,7 @@
       ]
     },
     "app.factory": {
-      "loaded_app_module_count": 552,
+      "loaded_app_module_count": 553,
       "max_ms": 1127.911,
       "median_ms": 1122.382,
       "min_ms": 1119.221,
@@ -28,7 +28,7 @@
       ]
     },
     "app.main": {
-      "loaded_app_module_count": 554,
+      "loaded_app_module_count": 555,
       "max_ms": 1188.652,
       "median_ms": 1183.509,
       "min_ms": 1174.522,

tests/test_history_ai_retry_gate.py

@@ -3,11 +3,15 @@
 from __future__ import annotations
 
 import asyncio
+from unittest.mock import Mock
+
+import pytest
 
 from app.api.endpoints import history as history_endpoint
 from app.application.configuration import ApiRuntimeConfig
 from app.application.transfer.execution import (
     TransferExecutionState,
+    TransferFailureDiscardResult,
     TransferRetryRequestResult,
 )
 from app.runtime.progress import AsyncProgressHelper
@@ -368,3 +372,50 @@ def test_batch_ai_redo_sends_only_legacy_records_after_durable_acceptance(
     assert response.data["history_ids"] == [24, 25]
     assert prompted == [[25]]
     assert started[0]["history_ids"] == [25]
+
+
+@pytest.mark.parametrize("discarded", [True, False])
+def test_discard_corrupt_history_returns_typed_result(discarded):
+    """清理端点保留历史 ID，并透传持久层的清理成功或拒绝结果。"""
+    repository = Mock()
+    repository.discard_corrupt_by_history.return_value = TransferFailureDiscardResult(
+        discarded=discarded, state=None, message="清理结果",
+    )
+    response = history_endpoint.discard_corrupt_transfer_history(
+        history_id=7,
+        history=TransferHistory(id=7, transfer_task_id="task-7"),
+        execution_repository=repository,
+        _=None,
+    )
+    assert response.success is discarded
+    assert response.message == "清理结果"
+    assert response.data.model_dump() == {"history_id": 7}
+    repository.discard_corrupt_by_history.assert_called_once_with(task_id="task-7", history_id=7)
+
+
+@pytest.mark.parametrize("exists", [True, False])
+def test_discard_corrupt_history_without_task_is_idempotent(exists):
+    """已清理历史返回同一结构，缺失历史返回失败且均不调用持久层清理。"""
+    repository = Mock()
+    response = history_endpoint.discard_corrupt_transfer_history(
+        history_id=7,
+        history=TransferHistory(id=7) if exists else None,
+        execution_repository=repository,
+        _=None,
+    )
+    assert response.success is exists
+    if exists:
+        assert response.data.model_dump() == {"history_id": 7}
+        assert response.message == "整理任务已清理"
+    else:
+        assert response.data is None
+        assert response.message == "整理记录不存在"
+    repository.discard_corrupt_by_history.assert_not_called()
+
+
+def test_discard_corrupt_history_dependency_loads_dto():
+    """同步清理前通过异步依赖读取历史，避免在事件循环运行同步写事务。"""
+    result = asyncio.run(history_endpoint._get_discard_transfer_history(
+        history_id=7, query=_HistoryQuery([TransferHistory(id=7)]),
+    ))
+    assert result.id == 7

tests/test_site_cookie_endpoint.py

@@ -51,7 +51,7 @@ def test_update_cookie_legacy_get_keeps_query_params():
         )
 
     assert response.success is False
-    assert response.message == "操作失败，请稍后重试"
+    assert response.message == "failed"
     fake_chain.update_cookie.assert_called_once_with(
         site_info=fake_site,
         username="user",

tests/test_subscription_execution_status.py

@@ -132,15 +132,15 @@ def test_execution_status_exposes_scheduled_new_search_without_failure():
     assert statuses[4].error is None
 
 
-def test_failed_search_exposes_safe_error():
-    """搜索失败文本必须压平且不暴露内部错误细节。"""
+def test_failed_search_preserves_normalized_business_error():
+    """搜索失败文本必须压平空白并保留业务层给出的失败原因。"""
     repository = _Repository()
     repository.tasks[3] = _task(3, state="failed", phase="failed")
 
     statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((3,)))
 
     assert statuses[3].state == "failed"
-    assert statuses[3].error == "订阅操作失败，请刷新后重试"
+    assert statuses[3].error == "provider timeout"
 
 
 def test_batch_requires_complete_subscription_access():

tests/test_transfer_durable_retry_owner.py

@@ -79,7 +79,7 @@ def _install_discard_port(monkeypatch) -> object:
         message="已放弃这条失败的整理任务",
     )
     monkeypatch.setattr(
-        "app.chain.transfer.records.TransferExecutionCommand",
+        "app.chain.transfer.records.TransferRecoveryCommand",
         _DiscardCommand,
     )
     return repository

tests/test_transfer_execution_migration.py

@@ -779,7 +779,7 @@ def test_migrated_legacy_reviews_are_discoverable_resolvable_and_retryable(
 
     with pytest.raises(
             TransferExecutionConflictError,
-            match="没有足够证据证明外部操作已发生",
+            match="这条整理步骤无法确认是否已经执行，请人工确认文件状态后再继续",
     ):
         command.resolve_manual_review(
             task_id="planned",
@@ -840,7 +840,7 @@ def test_migrated_legacy_reviews_are_discoverable_resolvable_and_retryable(
             assert pending.state in {"accepted", "planned"}
         with pytest.raises(
                 TransferExecutionConflictError,
-                match="可执行规划状态|完整计划检查点|无法恢复",
+                match="整理任务记录不完整，请重新识别文件后再整理|整理任务尚未准备完成，请稍后再试",
         ):
             command.prepare(
                 task_id=task_id,

tests/test_transfer_execution_persistence.py

@@ -22,6 +22,7 @@ from app.application.transfer.execution import (
     build_transfer_checkpoint_fingerprint,
     build_transfer_operation_id,
 )
+from app.application.transfer.recovery import TransferRecoveryCommand
 from app.application.transfer.workflow import (
     TransferPlanCheckpoint,
     TransferPlanItem,
@@ -888,7 +889,8 @@ def test_discard_failed_removes_execution_evidence_and_detaches_history(
 ) -> None:
     """放弃匹配的 FAILED 任务应删除执行证据，并把历史恢复为普通记录。"""
     history_id = _seed_failed_receipt(execution_store)
-    _, command = _repository(execution_store)
+    repository, _ = _repository(execution_store)
+    command = TransferRecoveryCommand(repository)
 
     result = command.discard_failed(
         task_id="task-1",
@@ -918,7 +920,8 @@ def test_discard_failed_rejects_nonfailed_execution_state(
         execution_store,
         execution_state=execution_state,
     )
-    _, command = _repository(execution_store)
+    repository, _ = _repository(execution_store)
+    command = TransferRecoveryCommand(repository)
 
     result = command.discard_failed(
         task_id="task-1",
@@ -939,7 +942,8 @@ def test_discard_failed_rejects_nonfailed_execution_state(
 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)
+    repository, _ = _repository(execution_store)
+    command = TransferRecoveryCommand(repository)
 
     result = command.discard_failed(
         task_id="task-1",

tests/test_transfer_failed_retry_buttons.py

@@ -8,7 +8,7 @@ import pytest
 
 from app.application.messaging.interaction import InteractionContext
 from app.chain.message import MessageChain
-from app.chain.transfer import TransferChain
+from app.chain.transfer.facade import TransferChain
 from app.runtime.config import settings
 from app.runtime.loop import main_loop_registry
 from app.runtime.tasks import TaskRegistry
@@ -26,11 +26,12 @@ def replace_main_loop() -> Callable[[object], None]:
 
 
 def test_build_failed_transfer_buttons():
-    """整理失败消息应提供重试与智能助手接管按钮。"""
+    """整理失败消息应提供重试、重新生成计划与智能助手接管按钮。"""
     buttons = TransferChain.build_failed_transfer_buttons(12)
 
     assert buttons == [[
         {"text": "重试", "callback_data": "transfer_retry_12"},
+        {"text": "重新生成计划", "callback_data": "transfer_regenerate_12"},
         {
             "text": "智能助手接管",
             "callback_data": "transfer_ai_retry_12",
@@ -134,12 +135,15 @@ def test_transfer_ai_retry_callback_schedules_agent_takeover(replace_main_loop):
     async_messages = []
 
     def _run_pending_coro(coro, *args, **kwargs):
+        """在测试线程执行协程，避免提交真实后台任务。"""
         asyncio.run(coro)
 
     async def _capture_message(message):
+        """记录异步消息以核对接管反馈。"""
         async_messages.append(message)
 
     async def _finish_immediately(**kwargs):
+        """立即回调完成结果，避免启动真实 Agent。"""
         kwargs["output_callback"]("ok")
 
     manager = SimpleNamespace(run_background_prompt=_finish_immediately)
@@ -286,16 +290,19 @@ def test_transfer_ai_retry_callback_uses_successful_move_dest_as_source(
     )
 
     def _run_pending_coro(coro, *args, **kwargs):
+        """同步执行协程并返回调度占位结果。"""
         asyncio.run(coro)
         return SimpleNamespace()
 
     async def fake_run_background_prompt(**kwargs):
+        """记录接管提示并回调模拟完成结果。"""
         captured["message"] = kwargs["message"]
         output_callback = kwargs.get("output_callback")
         if output_callback:
             output_callback("ok")
 
     async def fake_async_post_message(*args, **kwargs):
+        """隔离通知发送，避免测试访问真实消息渠道。"""
         return None
 
     from app.agent.prompt.transfer import build_manual_redo_prompt

tests/test_transfer_recovery_command.py

@@ -0,0 +1,63 @@
+"""整理恢复用例的清理边界与错误证据保留测试。"""
+
+from unittest.mock import Mock
+
+import pytest
+
+from app.application.transfer.execution import TransferExecutionConflictError
+from app.application.transfer.recovery import TransferRecoveryCommand
+from app.chain.transfer.settlement import _discard_corrupt_transfer_task
+
+
+@pytest.mark.parametrize("message", ["记录已失效", "记录不完整", "版本不一致", "检查点", "恢复状态不完整"])
+def test_corrupt_conflict_discards_owned_task(message):
+    """不可恢复的规划冲突必须带当前租约原子清理持久任务。"""
+    repository = Mock()
+    repository.discard_corrupt_task.return_value = True
+    result = TransferRecoveryCommand(repository).discard_conflict(
+        task_id="task", lease_token="lease", error=TransferExecutionConflictError(message),
+    )
+    assert result is True
+    repository.discard_corrupt_task.assert_called_once_with(
+        task_id="task", lease_token="lease", error=message,
+    )
+
+
+@pytest.mark.parametrize("error,task_id,lease_token", [
+    (ValueError("记录不完整"), "task", "lease"),
+    (TransferExecutionConflictError("暂时失败"), "task", "lease"),
+    (TransferExecutionConflictError("记录不完整"), None, "lease"),
+    (TransferExecutionConflictError("记录不完整"), "task", None),
+])
+def test_recoverable_or_unowned_failure_preserves_evidence(error, task_id, lease_token):
+    """普通错误、可重试冲突和缺少租约的任务都不得删除恢复证据。"""
+    repository = Mock()
+    assert TransferRecoveryCommand(repository).discard_conflict(
+        task_id=task_id, lease_token=lease_token, error=error,
+    ) is False
+    repository.discard_corrupt_task.assert_not_called()
+
+
+@pytest.mark.parametrize("preview", [True, False])
+def test_cleanup_preserves_preview_and_tolerates_repository_failure(preview):
+    """预览不清理任务；持久清理失败不阻止内存作业收口。"""
+    repository = Mock()
+    repository.discard_corrupt_task.side_effect = RuntimeError("存储失败")
+    task = Mock(preview=preview, admission_task_id="task", lease_token="lease")
+    _discard_corrupt_transfer_task(repository, task, TransferExecutionConflictError("记录不完整"))
+    assert repository.discard_corrupt_task.call_count == (0 if preview else 1)
+
+
+@pytest.mark.parametrize("method,kwargs", [
+    ("discard_corrupt_task", {"task_id": "", "lease_token": "lease", "error": "损坏"}),
+    ("discard_corrupt_task", {"task_id": "task", "lease_token": "", "error": "损坏"}),
+    ("discard_corrupt_task", {"task_id": "task", "lease_token": "lease", "error": ""}),
+    ("discard_corrupt_by_history", {"task_id": "task", "history_id": 0}),
+    ("discard_failed", {"task_id": "task", "history_id": 7, "settlement_revision": 0}),
+])
+def test_invalid_cleanup_identity_never_reaches_repository(method, kwargs):
+    """缺少身份或结算证据时，必须在写持久层之前拒绝清理。"""
+    repository = Mock()
+    with pytest.raises(ValueError):
+        getattr(TransferRecoveryCommand(repository), method)(**kwargs)
+    assert repository.mock_calls == []