From a69c9073cb50d89f3888c5d6db9aa16bd70c5329 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 8 Sep 2026 23:15:38 +0800 Subject: [PATCH] feat(transfer): implement recovery commands for discarding corrupt tasks --- app/api/endpoints/history.py | 37 +++++++--- app/application/historymutation.py | 4 +- app/application/plugin/data.py | 3 +- app/application/subscription/execution.py | 3 +- app/application/transfer/execution.py | 43 ----------- app/application/transfer/recovery.py | 73 +++++++++++++++++++ app/chain/download/subtitle.py | 3 +- app/chain/transfer/records.py | 4 +- app/chain/transfer/settlement.py | 39 ++++------ app/locales/en-US.json | 1 + app/locales/zh-CN.json | 1 + app/locales/zh-TW.json | 1 + app/schemas/exports.py | 1 + app/schemas/history.py | 6 ++ docs/architecture-overview.md | 4 +- .../architecture/agent-api-surface-audit.json | 16 +++- docs/architecture/agent-api-surface-audit.md | 5 +- docs/architecture/optimization-checklist.md | 2 +- docs/mcp-api.md | 16 ++-- docs/rules/05-architecture.md | 2 +- scripts/generate_agent_api_surface_audit.py | 7 ++ skills/moviepilot-api/SKILL.md | 4 + .../architecture/dependency-baseline.json | 14 +++- .../startup-performance-baseline.json | 6 +- tests/test_history_ai_retry_gate.py | 51 +++++++++++++ tests/test_site_cookie_endpoint.py | 2 +- tests/test_subscription_execution_status.py | 6 +- tests/test_transfer_durable_retry_owner.py | 2 +- tests/test_transfer_execution_migration.py | 4 +- tests/test_transfer_execution_persistence.py | 10 ++- tests/test_transfer_failed_retry_buttons.py | 11 ++- tests/test_transfer_recovery_command.py | 63 ++++++++++++++++ 32 files changed, 327 insertions(+), 117 deletions(-) create mode 100644 app/application/transfer/recovery.py create mode 100644 tests/test_transfer_recovery_command.py diff --git a/app/api/endpoints/history.py b/app/api/endpoints/history.py index 957883182..7b9e60cc8 100644 --- a/app/api/endpoints/history.py +++ b/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( diff --git a/app/application/historymutation.py b/app/application/historymutation.py index 41ee9b0df..d2ad61c3d 100644 --- a/app/application/historymutation.py +++ b/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, diff --git a/app/application/plugin/data.py b/app/application/plugin/data.py index 48ced04d3..f6141a8ad 100644 --- a/app/application/plugin/data.py +++ b/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 diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index a1fbdc45f..2ebc95e13 100644 --- a/app/application/subscription/execution.py +++ b/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 diff --git a/app/application/transfer/execution.py b/app/application/transfer/execution.py index cf1f0aa6a..1028d5e60 100644 --- a/app/application/transfer/execution.py +++ b/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, *, diff --git a/app/application/transfer/recovery.py b/app/application/transfer/recovery.py new file mode 100644 index 000000000..ca5b9c783 --- /dev/null +++ b/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, + ) diff --git a/app/chain/download/subtitle.py b/app/chain/download/subtitle.py index ccc88788b..3b668d177 100644 --- a/app/chain/download/subtitle.py +++ b/app/chain/download/subtitle.py @@ -1,6 +1,7 @@ -from __future__ import annotations """字幕获取、解压和存储 owner。""" +from __future__ import annotations + import re import shutil import time diff --git a/app/chain/transfer/records.py b/app/chain/transfer/records.py index ac0dfeb39..795648dbf 100644 --- a/app/chain/transfer/records.py +++ b/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, diff --git a/app/chain/transfer/settlement.py b/app/chain/transfer/settlement.py index 7301be039..0d16bc60d 100644 --- a/app/chain/transfer/settlement.py +++ b/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) diff --git a/app/locales/en-US.json b/app/locales/en-US.json index 40213b0d7..15e1ed805 100644 --- a/app/locales/en-US.json +++ b/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", diff --git a/app/locales/zh-CN.json b/app/locales/zh-CN.json index f5be3af97..0db239063 100644 --- a/app/locales/zh-CN.json +++ b/app/locales/zh-CN.json @@ -103,6 +103,7 @@ } }, "messages": { + "整理任务已清理": "整理任务已清理", "当前管理用户缺少可审计身份": "当前管理用户缺少可审计身份", "人工复核任务不存在": "人工复核任务不存在", "媒体来源和媒体 ID 必须同时提供": "媒体来源和媒体 ID 必须同时提供", diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index eb8e4b422..e7171a5b2 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -103,6 +103,7 @@ } }, "messages": { + "整理任务已清理": "整理任務已清理", "暂无到期订阅，无需搜索": "暫無到期訂閱，無需搜尋", "调用工具失败": "調用工具失敗", "媒体来源和媒体 ID 必须同时提供": "媒體來源和媒體 ID 必須同時提供", diff --git a/app/schemas/exports.py b/app/schemas/exports.py index 4c17c12a1..b6380ac2d 100644 --- a/app/schemas/exports.py +++ b/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'), diff --git a/app/schemas/history.py b/app/schemas/history.py index 2078ee739..d25c91567 100644 --- a/app/schemas/history.py +++ b/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): """整理历史分页数据。""" diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index be0b15e43..bf474bf7d 100644 --- a/docs/architecture-overview.md +++ b/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） | diff --git a/docs/architecture/agent-api-surface-audit.json b/docs/architecture/agent-api-surface-audit.json index 19d633d75..fdabf3114 100644 --- a/docs/architecture/agent-api-surface-audit.json +++ b/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", diff --git a/docs/architecture/agent-api-surface-audit.md b/docs/architecture/agent-api-surface-audit.md index 93d40bf1b..c8b49a004 100644 --- a/docs/architecture/agent-api-surface-audit.md +++ b/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 | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 855e1e649..4247219e5 100644 --- a/docs/architecture/optimization-checklist.md +++ b/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 等基线均为零 | diff --git a/docs/mcp-api.md b/docs/mcp-api.md index c5e2b4678..bf440dc3d 100644 --- a/docs/mcp-api.md +++ b/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 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index d2e1c45af..81c70262e 100644 --- a/docs/rules/05-architecture.md +++ b/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 | diff --git a/scripts/generate_agent_api_surface_audit.py b/scripts/generate_agent_api_surface_audit.py index 577fd3d74..9fb060208 100644 --- a/scripts/generate_agent_api_surface_audit.py +++ b/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", diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index eea38d4b0..787d2278d 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/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: diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 95ed558bc..1b951304e 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/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", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index 6526f9936..103cb578a 100644 --- a/tests/fixtures/architecture/startup-performance-baseline.json +++ b/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, diff --git a/tests/test_history_ai_retry_gate.py b/tests/test_history_ai_retry_gate.py index f98d5b16b..227a3ca0c 100644 --- a/tests/test_history_ai_retry_gate.py +++ b/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 diff --git a/tests/test_site_cookie_endpoint.py b/tests/test_site_cookie_endpoint.py index b8959ad3f..84bc3fcee 100644 --- a/tests/test_site_cookie_endpoint.py +++ b/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", diff --git a/tests/test_subscription_execution_status.py b/tests/test_subscription_execution_status.py index 06aee2788..e808bc21b 100644 --- a/tests/test_subscription_execution_status.py +++ b/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(): diff --git a/tests/test_transfer_durable_retry_owner.py b/tests/test_transfer_durable_retry_owner.py index d00e80419..91927b943 100644 --- a/tests/test_transfer_durable_retry_owner.py +++ b/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 diff --git a/tests/test_transfer_execution_migration.py b/tests/test_transfer_execution_migration.py index 84888a533..73b8b437b 100644 --- a/tests/test_transfer_execution_migration.py +++ b/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, diff --git a/tests/test_transfer_execution_persistence.py b/tests/test_transfer_execution_persistence.py index 6c66920ee..065e6b8b2 100644 --- a/tests/test_transfer_execution_persistence.py +++ b/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", diff --git a/tests/test_transfer_failed_retry_buttons.py b/tests/test_transfer_failed_retry_buttons.py index 730d6c96e..00521afd7 100644 --- a/tests/test_transfer_failed_retry_buttons.py +++ b/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 diff --git a/tests/test_transfer_recovery_command.py b/tests/test_transfer_recovery_command.py new file mode 100644 index 000000000..c5ae4d554 --- /dev/null +++ b/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 == []