fix(api): sync v3 and complete agent media type compatibility

2026-11-05 22:59:46 +08:00 · 2026-10-10 12:22:57 +08:00
parent 0d62815a0f bb29451222
commit cdb8d9df71
15 changed files with 490 additions and 21 deletions
--- a/app/api/endpoints/douban.py
+++ b/app/api/endpoints/douban.py
@@ -68,7 +68,7 @@ async def douban_credits(
    count: CompatibleCountParam = None,
 ) -> Any:
    """
-    根据豆瓣ID查询演员阵容，type_name: 电影/电视剧
+    根据豆瓣ID查询演员阵容，type_name: 电影/movie、电视剧/tv
    """
    mediatype = MediaType.from_agent(type_name) or MediaType(type_name)
    if mediatype == MediaType.MOVIE:
@@ -91,7 +91,7 @@ async def douban_recommend(
    _: _SchemaTokenPayload = Depends(verify_token),
 ) -> Any:
    """
-    根据豆瓣ID查询推荐电影/电视剧，type_name: 电影/电视剧
+    根据豆瓣ID查询推荐电影/电视剧，type_name: 电影/movie、电视剧/tv
    """
    mediatype = MediaType.from_agent(type_name) or MediaType(type_name)
    if mediatype == MediaType.MOVIE:
--- a/app/api/endpoints/mediaserver.py
+++ b/app/api/endpoints/mediaserver.py
@@ -162,7 +162,7 @@ def not_exists(
    count: CompatibleCountParam = None,
 ) -> Any:
    """
-    根据媒体信息查询缺失电影/剧集
+    根据媒体信息查询缺失电影/剧集，媒体类型兼容英文与中文值
    """
    # 媒体信息
    meta = MetaInfo(title=media_in.title)
@@ -176,7 +176,8 @@ def not_exists(
        meta.year = media_in.year
    # 转化为媒体信息对象
    mediainfo = MediaInfo()
-    mediainfo.from_dict(media_in.model_dump())
+    # 复用已解析的类型，避免领域对象再次将 Agent 英文值按中文枚举解析。
+    mediainfo.from_dict({**media_in.model_dump(), "type": mtype})
    exist_flag, no_exists = DownloadChain().get_no_exists_info(
        meta=meta, mediainfo=mediainfo
    )
--- a/app/api/endpoints/tmdb.py
+++ b/app/api/endpoints/tmdb.py
@@ -114,7 +114,7 @@ async def tmdb_similar(
    _: _SchemaTokenPayload = Depends(verify_token),
 ) -> Any:
    """
-    根据TMDBID查询类似电影/电视剧，type_name: 电影/电视剧
+    根据TMDBID查询类似电影/电视剧，type_name: 电影/movie、电视剧/tv
    """
    mediatype = MediaType.from_agent(type_name) or MediaType(type_name)
    if mediatype == MediaType.MOVIE:
@@ -141,7 +141,7 @@ async def tmdb_recommend(
    _: _SchemaTokenPayload = Depends(verify_token),
 ) -> Any:
    """
-    根据TMDBID查询推荐电影/电视剧，type_name: 电影/电视剧
+    根据TMDBID查询推荐电影/电视剧，type_name: 电影/movie、电视剧/tv
    """
    mediatype = MediaType.from_agent(type_name) or MediaType(type_name)
    if mediatype == MediaType.MOVIE:
@@ -187,7 +187,7 @@ async def tmdb_credits(
    _: _SchemaTokenPayload = Depends(verify_token),
 ) -> Any:
    """
-    根据TMDBID查询演员阵容，type_name: 电影/电视剧
+    根据TMDBID查询演员阵容，type_name: 电影/movie、电视剧/tv
    """
    mediatype = MediaType.from_agent(type_name) or MediaType(type_name)
    if mediatype == MediaType.MOVIE:
--- a/app/api/endpoints/transfer.py
+++ b/app/api/endpoints/transfer.py
@@ -791,14 +791,12 @@ def _execute_manual_transfer(
    type_name = str(transer_item.type_name).strip() if transer_item.type_name else ""
    if type_name and type_name.lower() not in {"自动", "auto", "none"}:
        # 先按 Agent 使用的 movie/tv/music 取值解析，再回退到原生中文枚举值。
-        mtype = MediaType.from_agent(type_name)
-        if mtype is None:
-            try:
-                mtype = MediaType(type_name)
-            except ValueError:
-                return _SchemaResponse(
-                    success=False, message=f"不支持的媒体类型：{type_name}"
-                )
+        try:
+            mtype = MediaType.from_agent(type_name) or MediaType(type_name)
+        except ValueError:
+            return _SchemaResponse(
+                success=False, message=f"不支持的媒体类型：{type_name}"
+            )

    def _resolve_music_type(file_item: FileItem) -> Optional[str]:
        """为未显式指定实体的旧客户端按源项类型补全音乐命名空间。"""
--- a/app/application/transfer/execution.py
+++ b/app/application/transfer/execution.py
@@ -508,6 +508,11 @@ class TransferExecutionRepository(Protocol):
    def get_snapshot(self, *, task_id: str) -> Optional[TransferExecutionSnapshot]:
        """读取任务执行快照。"""

+    def isolate_unhandled_failure(
+            self, *, task_id: str, lease_token: str, error: str,
+    ) -> bool:
+        """将步骤账本之外的异常计入失败次数并隔离复核，不推测外部操作结果。"""
+
    async def async_get_snapshot(
            self,
            *,
--- a/app/chain/transfer/execution.py
+++ b/app/chain/transfer/execution.py
@@ -14,6 +14,8 @@ from app.application.history.retry import (
 from app.application.transfer.execution import (
    TransferExecutionCheckpoint,
    TransferExecutionCommand,
+    TransferExecutionConflictError,
+    TransferExecutionLeaseLostError,
    TransferExecutionRepository,
    TransferExecutionSnapshot,
    TransferOperationObservation,
@@ -313,7 +315,20 @@ class TransferExecutionOwner(_TransferOwnerBase):
            *,
            preview: bool,
    ) -> Tuple[bool, str]:
-        """收口整理执行异常；人工复核只告警并保留 durable 状态。"""
+        """在插件可替换执行入口之外收口异常，防止未记账失败无限回放。"""
+        if (not preview and not task.terminal_settled and task.admission_task_id and task.lease_token
+                and not isinstance(error, (TransferAdmissionConflictError, TransferLeaseLostError,
+                                           TransferExecutionConflictError, TransferExecutionLeaseLostError,
+                                           _TransferRetryDeferred, _TransferRetryExhausted,
+                                           _TransferManualReviewRequired))):
+            try:
+                isolated = self.transfer_execution_repository.isolate_unhandled_failure(
+                    task_id=task.admission_task_id, lease_token=task.lease_token, error=str(error),
+                )
+                if isolated:
+                    error = _TransferManualReviewRequired(str(error))
+            except Exception as record_error:
+                logger.error(f"记录整理执行异常失败：{record_error}")
        if isinstance(error, _TransferManualReviewRequired):
            logger.warning(
                f"{task.fileitem.name} 已转入人工复核，未自动重放；"
--- a/app/db/adapters/transfer/execution.py
+++ b/app/db/adapters/transfer/execution.py
@@ -55,6 +55,10 @@ from app.db.oper.transferpending import TransferPendingOper
 from app.db.uow import SqlAlchemyUnitOfWork

 _LEGACY_REVIEW_STEP_KIND = "legacy_execution_review"
+_UNHANDLED_REVIEW_STEP_KIND = "unhandled_execution_review"
+_REVIEW_STEP_KINDS = {_LEGACY_REVIEW_STEP_KIND, _UNHANDLED_REVIEW_STEP_KIND}
+# 与迁移保留的最大序号相邻，复核凭据不占用从零连续递增的真实执行序号。
+_UNHANDLED_REVIEW_STEP_ORDINAL = 2_147_483_646


 def _local_now() -> datetime:
@@ -100,8 +104,61 @@ class TransactionalTransferExecutionRepository:
    def _execution_steps(
            steps: list[TransferExecutionStepModel],
    ) -> list[TransferExecutionStepModel]:
-        """排除只用于迁移人工审计、不得参与真实执行的 synthetic 步骤。"""
-        return [step for step in steps if step.kind != _LEGACY_REVIEW_STEP_KIND]
+        """排除只用于人工审计、不得参与真实执行的 synthetic 步骤。"""
+        return [step for step in steps if step.kind not in _REVIEW_STEP_KINDS]
+
+    def isolate_unhandled_failure(
+            self, *, task_id: str, lease_token: str, error: str,
+    ) -> bool:
+        """原子隔离未记账异常并保存可发现的复核凭据，保留所有真实步骤证据。"""
+        now_utc, updated_at = self._times()
+        with self._session_factory() as session:
+            transaction = SqlAlchemyUnitOfWork(session)
+            try:
+                pending_oper = TransferPendingOper(session)
+                updated = pending_oper.stage_isolate_unhandled_failure(
+                    task_id=task_id, lease_token=lease_token, error=error,
+                    now_utc=now_utc, updated_at=updated_at,
+                )
+                if updated != 1:
+                    transaction.commit()
+                    return False
+                session.expire_all()
+                pending = pending_oper.get_by_task_id(task_id=task_id)
+                if pending is None:
+                    raise TransferExecutionConflictError("异常隔离后未找到整理任务")
+                self._stage_unhandled_review(session, pending=pending, error=error, updated_at=updated_at)
+                transaction.commit()
+                return True
+            except Exception:
+                self._rollback(transaction)
+                raise
+
+    @staticmethod
+    def _stage_unhandled_review(
+            session: Session, *, pending: TransferPending, error: str, updated_at: str,
+    ) -> None:
+        """创建或更新独立复核凭据；同一任务再次异常时复用凭据并保留复核审计。"""
+        oper = TransferExecutionStepOper(session)
+        step = next((item for item in oper.list_by_task_id(task_id=pending.task_id)
+                     if item.kind == _UNHANDLED_REVIEW_STEP_KIND), None)
+        if step is None:
+            intent = TransferStepIntent.create(
+                task_id=pending.task_id,
+                checkpoint_fingerprint=pending.input_fingerprint,
+                ordinal=_UNHANDLED_REVIEW_STEP_ORDINAL,
+                phase="review", kind=_UNHANDLED_REVIEW_STEP_KIND,
+                payload={"source": {"storage": pending.storage, "path": pending.src_path}},
+            )
+            step = oper.stage_prepare(
+                task_id=pending.task_id, operation_id=intent.operation_id,
+                checkpoint_fingerprint=intent.checkpoint_fingerprint,
+                ordinal=intent.ordinal, phase=intent.phase, kind=intent.kind,
+                intent_version=intent.version, intent_payload=dict(intent.payload), now_time=updated_at,
+            )
+        step.state = TransferStepState.MANUAL_REVIEW.value
+        step.last_error = error
+        step.updated_at = updated_at

    @staticmethod
    def _project_step(step: TransferExecutionStepModel) -> TransferExecutionStep:
@@ -1533,7 +1590,7 @@ class TransactionalTransferExecutionRepository:
                        decision is TransferManualReviewDecision.APPLIED
                        and step is not None
                        and step.task_id == task_id
-                        and step.kind == _LEGACY_REVIEW_STEP_KIND
+                        and step.kind in _REVIEW_STEP_KINDS
                ):
                    raise TransferExecutionConflictError(
                        "这条整理步骤无法确认是否已经执行，请人工确认文件状态后再继续；"
--- a/app/db/models/transferpending.py
+++ b/app/db/models/transferpending.py
@@ -1127,6 +1127,37 @@ class TransferPending(Base):
            execution_options={"synchronize_session": False},
        )

+    @classmethod
+    def isolate_unhandled_failure(
+            cls, db: Session, *, task_id: str, lease_token: str,
+            error: str, now_utc: str, updated_at: str,
+    ) -> int:
+        """以当前租约收口未记账异常；已释放租约的重试和复核不得重复计数。"""
+        return execute_dml(
+            db,
+            update(cls).where(
+                cls.task_id == task_id,
+                cls.state.in_(("accepted", "planned", "provider_pending")),
+                cls.execution_state.in_(("not_started", "running", "retry_wait")),
+                cls.execution_payload.is_(None),
+                cls.terminal_history_id.is_(None),
+                cls.lease_token == lease_token,
+                cls.lease_expires_at.is_not(None),
+                cls.lease_expires_at > now_utc,
+            ).values(
+                execution_state="manual_review",
+                retry_count=cls.retry_count + 1,
+                retry_due_at=None,
+                lease_owner=None,
+                lease_token=None,
+                lease_expires_at=None,
+                heartbeat_at=None,
+                last_error=error,
+                updated_at=updated_at,
+            ),
+            execution_options={"synchronize_session": False},
+        )
+
    @classmethod
    def record_planning_failure(cls, db: Session, *, task_id: str,
                                lease_token: str, error: str,
--- a/app/db/oper/transferpending.py
+++ b/app/db/oper/transferpending.py
@@ -12,6 +12,18 @@ class TransferPendingOper(DbOper):
    整理链的文件重新送回去，避免挂载故障重启后永久漏件。
    """

+    def stage_isolate_unhandled_failure(
+            self, *, task_id: str, lease_token: str, error: str,
+            now_utc: str, updated_at: str,
+    ) -> int:
+        """在调用方事务中暂存异常隔离，失败计数与租约释放使用同一 CAS。"""
+        return self._execute_sync_write(
+            lambda session: TransferPending.isolate_unhandled_failure(
+                session, task_id=task_id, lease_token=lease_token,
+                error=error, now_utc=now_utc, updated_at=updated_at,
+            )
+        )
+
    def stage_admit(self, *, task_id: str, storage: str, src_path: str,
                    state: str, now_time: str, input_version: int,
                    planning_input: dict[str, Any],
--- a/docs/mcp-api.md
+++ b/docs/mcp-api.md
@@ -419,6 +419,8 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返

 媒体来源列表 `/api/v1/media/source` 仅预置上述九个来源，其余来源由启用插件注册后提供。哔哩哔哩、芒果 TV、咪咕视频、腾讯视频、爱奇艺不再占用内置来源标识，宿主也不再转换这些插件来源的旧别名；调用方应使用插件声明的准确来源 ID。

+媒体详情的 `type_name`、新增订阅和媒体库缺失查询的 `type`、手动整理的 `type_name` 均兼容 `movie`、`tv`、`music` 及对应中文值 `电影`、`电视剧`、`音乐`。TMDB 的相似、推荐和演员接口，以及豆瓣的推荐和演员接口同样兼容英文类型，但仍只提供电影和电视剧结果；传入音乐类型返回空列表。
+
 影视自动识别在未指定来源时只使用 TMDB，未命中时不会继续查询其它影视源。音乐路径识别严格按 AcoustID 音频指纹、文件标签、文件名三级依次执行；指纹或标签直接提供 MusicBrainz Recording ID 时，会直接查询 MusicBrainz 详情，未绑定身份的音乐标题及目录识别按 `SEARCH_SOURCE` 中内置音乐来源的配置顺序回退（MusicBrainz、TheAudioDB、豆瓣音乐），未选择音乐来源时兼容默认 MusicBrainz。显式 `media_source` / 主身份或 MusicBrainz 发行标签固定所属来源；来源之间不转换或拼接 ID。`MediaInfo` 响应仍可能包含 `tmdb_id`、`douban_id`、`bangumi_id`、`anilist_id` 等跨源映射辅助字段，但这些字段不是通用请求入口。明确归属 `/tmdb`、`/douban`、`/bangumi`、`/anilist` 的接口，以及固定使用 TMDB 的剧集组和排期接口，仍可按其单数据源契约接收原生 ID。

 | 方法 | 路径 | 说明 |
@@ -435,6 +437,11 @@ 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` 选填，省略或空白时记为空字符串，最多 2000 字符；`applied` 还必须提供 `result_payload`。`failed` 不属于公开决策，失败终态只能由持租约的 durable 结算写入；响应仅返回任务、操作、决策、后续状态和复核修订号 |

+步骤账本之外的未处理异常（例如插件替换整理入口后在 `finally` 抛错）会记一次失败，
+并立即进入 `manual_review` 停止自动回放。详情中的 `unhandled_execution_review` 是诊断凭据，
+不代表某个文件操作已执行；应先修复或关闭异常插件并确认文件状态，必要时人工回滚，
+再以 `not_applied` 交回调度器。该凭据不接受 `applied`，既有真实步骤仍按原证据恢复。
+
 #### SMB 下载器监控与服务端整理

 下载器原生保存路径与 MoviePilot 的存储访问路径是两个地址空间。下载器配置的
--- a/skills/moviepilot-api/SKILL.md
+++ b/skills/moviepilot-api/SKILL.md
@@ -1,6 +1,6 @@
 ---
 name: moviepilot-api
-version: 47
+version: 49
 description: >-
  Use this skill for MoviePilot product operations such as media search, torrent
  search, downloads, subscriptions, library checks, sites, storage, workflows,
@@ -67,6 +67,12 @@ allowed-api-operations: >-

 # MoviePilot API

+Unhandled exceptions outside the durable transfer step ledger stop automatic
+replay and create an `unhandled_execution_review` diagnostic. Repair or disable
+the failing plugin and inspect the files before resolving it as `not_applied`;
+roll back untracked effects first when needed. This diagnostic rejects `applied`
+and does not replace the evidence for any recorded file operations.
+
 For relevant prior conversations or previous tool evidence, use the host-provided
 `session_search` tool when available. Discover a session, then read its anchored
 messages; past results are context, not proof of current state or permission to
@@ -163,6 +169,10 @@ are separate operation contracts, not interchangeable parameter templates.
 subscription executions use `limit`, not `page` or `count`. `site.rss` lists
 RSS-enabled sites; it does not accept a `site_id` filter.

+`media.detail` and `transfer.file` accept `movie`, `tv`, or `music` in
+`type_name`; `subscription.add` accepts the same values in `type`. The
+corresponding Chinese labels remain supported.
+
 ## API Category Index

 Each category file contains the complete operation contracts for its namespace.
--- a/tests/test_api_media_types.py
+++ b/tests/test_api_media_types.py
@@ -0,0 +1,160 @@
+"""验证 API 将 Agent 英文类型与原生中文类型传入相同业务分支。"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from app.api.endpoints import douban, media, mediaserver, subscribe, tmdb, transfer
+from app.schemas.file import FileItem
+from app.schemas.subscribe import Subscribe
+from app.schemas.transfer import ManualTransferItem
+from app.schemas.types import MediaSource, MediaType
+from app.schemas.workflow import MediaInfo
+
+MEDIA_TYPES = [
+    ("movie", MediaType.MOVIE),
+    ("tv", MediaType.TV),
+    ("music", MediaType.MUSIC),
+    ("电影", MediaType.MOVIE),
+    ("电视剧", MediaType.TV),
+    ("音乐", MediaType.MUSIC),
+    (" MOVIE ", MediaType.MOVIE),
+    (" TV ", MediaType.TV),
+    (" MUSIC ", MediaType.MUSIC),
+]
+METADATA_ENDPOINTS = [
+    (douban, "DoubanChain", "douban_credits", "credits", "doubanid"),
+    (douban, "DoubanChain", "douban_recommend", "recommend", "doubanid"),
+    (tmdb, "TmdbChain", "tmdb_similar", "similar", "tmdbid"),
+    (tmdb, "TmdbChain", "tmdb_recommend", "recommend", "tmdbid"),
+    (tmdb, "TmdbChain", "tmdb_credits", "credits", "tmdbid"),
+]
+
+
+@pytest.mark.parametrize("type_name,expected", MEDIA_TYPES)
+@pytest.mark.parametrize("module,chain_name,endpoint,operation,id_key", METADATA_ENDPOINTS)
+def test_metadata_endpoints_route_media_types(
+    monkeypatch, type_name, expected, module, chain_name, endpoint, operation, id_key,
+) -> None:
+    """影视英文和中文类型选择相同方法，音乐类型保持空列表。"""
+    movie_call = AsyncMock(return_value=[])
+    tv_call = AsyncMock(return_value=[])
+    chain = SimpleNamespace(**{
+        f"async_movie_{operation}": movie_call,
+        f"async_tv_{operation}": tv_call,
+    })
+    monkeypatch.setattr(module, chain_name, lambda: chain)
+
+    result = asyncio.run(getattr(module, endpoint)(
+        **{id_key: "123" if id_key == "doubanid" else 123},
+        type_name=type_name,
+    ))
+
+    assert result == []
+    assert movie_call.await_count == int(expected == MediaType.MOVIE)
+    assert tv_call.await_count == int(expected == MediaType.TV)
+
+
+@pytest.mark.parametrize("type_name,expected", MEDIA_TYPES)
+def test_media_detail_forwards_normalized_type(monkeypatch, type_name, expected) -> None:
+    """详情入口应将三类中英文类型都转换成枚举再交给识别链。"""
+    recognize = AsyncMock(return_value=None)
+    monkeypatch.setattr(media, "MediaChain", lambda: SimpleNamespace(
+        async_recognize_media=recognize,
+    ))
+
+    asyncio.run(media.detail(
+        media_id="123", media_source=MediaSource.TMDB, type_name=type_name,
+    ))
+
+    recognize.assert_awaited_once_with(
+        media_source=MediaSource.TMDB, media_id="123", mtype=expected,
+    )
+
+
+@pytest.mark.parametrize("type_name,expected", MEDIA_TYPES)
+def test_create_subscription_forwards_normalized_type(monkeypatch, type_name, expected) -> None:
+    """新增订阅应兼容 Agent 类型，并向订阅链传入准确枚举。"""
+    add = AsyncMock(return_value=(1, ""))
+    monkeypatch.setattr(subscribe, "SubscribeChain", lambda: SimpleNamespace(async_add=add))
+
+    result = asyncio.run(subscribe.create_subscribe(
+        subscribe_in=Subscribe(name="测试媒体", type=type_name),
+        current_user=SimpleNamespace(name="admin", is_superuser=True),
+    ))
+
+    assert result.success is True
+    assert add.await_args.kwargs["mtype"] is expected
+
+
+@pytest.mark.parametrize("type_name,expected", MEDIA_TYPES)
+def test_popular_subscriptions_normalize_record_type(monkeypatch, type_name, expected) -> None:
+    """共享订阅记录使用英文类型时，公开响应仍输出原生中文类型。"""
+    statistic = AsyncMock(return_value=[{
+        "name": "测试媒体", "type": type_name, "count": 1,
+    }])
+    monkeypatch.setattr(
+        subscribe.MoviePilotServerHelper, "async_get_subscribe_statistic", statistic,
+    )
+
+    result = asyncio.run(subscribe.popular_subscribes(stype="all"))
+
+    assert len(result) == 1
+    assert result[0]["type"] == expected.value
+
+
+@pytest.mark.parametrize("type_name,expected", MEDIA_TYPES + [(None, MediaType.UNKNOWN)])
+def test_library_missing_forwards_meta_type(monkeypatch, type_name, expected) -> None:
+    """缺失查询应向下载链传入解析后的类型，空值沿用标题推断。"""
+    missing = Mock(return_value=(True, {}))
+    monkeypatch.setattr(mediaserver, "DownloadChain", lambda: SimpleNamespace(
+        get_no_exists_info=missing,
+    ))
+
+    assert mediaserver.not_exists(media_in=MediaInfo(title="测试电影", type=type_name)) == []
+    assert missing.call_args.kwargs["meta"].type is expected
+    assert missing.call_args.kwargs["mediainfo"].type is (
+        expected if type_name is not None else None
+    )
+
+
+@pytest.mark.parametrize("type_name,expected", MEDIA_TYPES + [
+    (None, None), ("", None), ("自动", None), ("auto", None), ("none", None),
+])
+def test_manual_transfer_forwards_normalized_type(monkeypatch, type_name, expected) -> None:
+    """手动整理兼容中英文类型，并保留未指定和自动识别语义。"""
+    execute = Mock(return_value=(True, ""))
+    monkeypatch.setattr(transfer, "TransferChain", lambda: SimpleNamespace(manual_transfer=execute))
+
+    result = transfer.manual_transfer(
+        transer_item=ManualTransferItem(
+            fileitem=FileItem(storage="local", path="/downloads/test.mkv", type="file"),
+            type_name=type_name,
+        ),
+        background=True,
+        history_query=Mock(),
+    )
+
+    assert result.success is True
+    assert execute.call_args.kwargs["mtype"] is expected
+
+
+def test_manual_transfer_rejects_invalid_type_before_execution(monkeypatch) -> None:
+    """非法媒体类型应保持失败响应，不能启动实际整理。"""
+    chain_factory = Mock()
+    monkeypatch.setattr(transfer, "TransferChain", chain_factory)
+
+    result = transfer.manual_transfer(
+        transer_item=ManualTransferItem(
+            fileitem=FileItem(storage="local", path="/downloads/test.mkv", type="file"),
+            type_name="unsupported",
+        ),
+        background=True,
+        history_query=Mock(),
+    )
+
+    assert result.success is False
+    assert result.message == "不支持的媒体类型：unsupported"
+    chain_factory.assert_not_called()
--- a/tests/test_builtin_skill_boundaries.py
+++ b/tests/test_builtin_skill_boundaries.py
@@ -117,7 +117,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None:
        "command-dispatch": "2",
        "database-operation": "9",
        "feedback-issue": "11",
-        "moviepilot-api": "47",
+        "moviepilot-api": "49",
        "moviepilot-update": "5",
        "organize-files": "5",
        "transfer-failed-retry": "5",
--- a/tests/test_transfer_execution_persistence.py
+++ b/tests/test_transfer_execution_persistence.py
@@ -14,6 +14,7 @@ from app.application.transfer.execution import (
    TransferExecutionConflictError,
    TransferExecutionState,
    TransferManualReviewDecision,
+    TransferManualReviewQuery,
    TransferOperationObservation,
    TransferOperationObservationState,
    TransferStepIntent,
@@ -220,6 +221,139 @@ def _intent(*, task_id: str = "task-1", ordinal: int = 0) -> TransferStepIntent:
    )


+@pytest.mark.parametrize("state", ["accepted", "planned"])
+@pytest.mark.parametrize("retry_count", [0, 318])
+@pytest.mark.parametrize("execution_state", ["not_started", "retry_wait"])
+def test_unhandled_failure_is_counted_once_and_stops_recovery(
+        execution_store, state, retry_count, execution_state,
+):
+    """补丁在步骤外抛异常时立即隔离，即使旧任务已反复领取也不再自动回放。"""
+    _seed_pending(execution_store, state=state, with_checkpoint=state == "planned")
+    with execution_store() as session:
+        pending = session.scalar(select(TransferPending))
+        pending.retry_count = retry_count
+        pending.attempt_count = 319
+        pending.execution_state = execution_state
+        session.commit()
+    repository, _ = _repository(execution_store)
+    error = "TransferChain has no attribute _TransferChain__finish_scrape_batch_task"
+    assert repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error=error)
+    assert not repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error=error)
+    snapshot = repository.get_snapshot(task_id="task-1")
+    assert snapshot.state is TransferExecutionState.MANUAL_REVIEW
+    assert snapshot.retry_count == retry_count + 1
+    assert snapshot.last_error == error
+    assert snapshot.steps == ()
+    view = TransferManualReviewQuery(repository).get(task_id="task-1")
+    assert view.step.kind == "unhandled_execution_review"
+    assert view.step.error == error
+    assert TransferManualReviewQuery(repository).list().total == 1
+    admissions = TransactionalTransferAdmissionRepository(execution_store)
+    assert admissions.claim_recoverable(owner_id="recovery", limit=10, lease_seconds=120) == []
+    with execution_store() as session:
+        pending = session.scalar(select(TransferPending))
+        assert pending.attempt_count == 319
+        assert pending.lease_owner is None
+        assert pending.lease_token is None
+        assert pending.lease_expires_at is None
+        assert pending.heartbeat_at is None
+        assert pending.retry_due_at is None
+
+
+@pytest.mark.parametrize("state,token,expires", [
+    ("retry_wait", None, "2099-01-01 00:00:00.000000"),
+    ("manual_review", "lease-1", "2099-01-01 00:00:00.000000"),
+    ("failed", "lease-1", "2099-01-01 00:00:00.000000"),
+    ("settling", "lease-1", "2099-01-01 00:00:00.000000"),
+    ("running", "new-owner-token", "2099-01-01 00:00:00.000000"),
+    ("running", "lease-1", "2000-01-01 00:00:00.000000"),
+])
+def test_unhandled_failure_cannot_overwrite_existing_disposition_or_lease(
+        execution_store, state, token, expires,
+):
+    """步骤已决定重试或终态、结算中及租约失效时，不重复计数或覆盖新 owner。"""
+    _seed_pending(execution_store)
+    with execution_store() as session:
+        pending = session.scalar(select(TransferPending))
+        pending.execution_state = state
+        pending.lease_token = token
+        pending.lease_expires_at = expires
+        pending.last_error = "previous evidence"
+        session.commit()
+    repository, _ = _repository(execution_store)
+    assert not repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer")
+    with execution_store() as session:
+        pending = session.scalar(select(TransferPending))
+        assert pending.execution_state == state
+        assert pending.retry_count == 0
+        assert pending.last_error == "previous evidence"
+        assert pending.lease_token == token
+        assert session.scalar(select(TransferExecutionStep)) is None
+
+
+def test_unhandled_failure_isolation_rolls_back_with_review_evidence(execution_store, monkeypatch):
+    """复核凭据写入失败时必须同时回滚状态、失败次数和租约释放。"""
+    _seed_pending(execution_store)
+    repository, _ = _repository(execution_store)
+
+    def fail_review(*_args, **_kwargs):
+        """模拟复核凭据无法写入。"""
+        raise RuntimeError("review write failed")
+
+    monkeypatch.setattr(repository, "_stage_unhandled_review", fail_review)
+    with pytest.raises(RuntimeError, match="review write failed"):
+        repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer")
+    with execution_store() as session:
+        pending = session.scalar(select(TransferPending))
+        assert pending.execution_state == "not_started"
+        assert pending.retry_count == 0
+        assert pending.lease_token == "lease-1"
+
+
+def test_unhandled_review_preserves_started_step_and_resumes_after_manual_confirmation(execution_store):
+    """复核未记账异常后，真实 STARTED 仍需严格探测，诊断步骤不参与执行或结算。"""
+    _seed_pending(execution_store)
+    repository, command = _repository(execution_store)
+    prepared = command.prepare(task_id="task-1", lease_token="lease-1", intent=_intent())
+    started = command.begin(task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id)
+    assert repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer")
+    snapshot = repository.get_snapshot(task_id="task-1")
+    assert snapshot.steps == (started,)
+    review = repository.get_manual_review(task_id="task-1")
+    with pytest.raises(TransferExecutionConflictError, match="人工回滚"):
+        command.resolve_manual_review(
+            task_id="task-1", operation_id=review.step.operation_id,
+            decision=TransferManualReviewDecision.APPLIED, actor="admin", reason="checked",
+            result=TransferStepResult(payload={"success": True}),
+        )
+    resolved = command.resolve_manual_review(
+        task_id="task-1", operation_id=review.step.operation_id,
+        decision=TransferManualReviewDecision.NOT_APPLIED, actor="admin", reason="plugin rolled back",
+    )
+    assert resolved.state is TransferExecutionState.RETRY_WAIT
+    assert repository.get_manual_review(task_id="task-1").review_revision == 1
+    admission = TransactionalTransferAdmissionRepository(execution_store).claim_task(
+        task_id="task-1", owner_id="recovery", lease_seconds=120,
+    )
+    command.prepare(task_id="task-1", lease_token=admission.lease_token, intent=_intent())
+    resumed = command.restart_after_not_applied(
+        task_id="task-1", lease_token=admission.lease_token, step=started,
+        evidence=TransferStepResult(payload={"observation": "not_applied"}),
+    )
+    completed = command.complete(
+        task_id="task-1", lease_token=admission.lease_token, step=resumed,
+        result=TransferStepResult(payload={"success": True}),
+    )
+    assert completed.state is TransferStepState.SUCCEEDED
+    assert repository.isolate_unhandled_failure(task_id="task-1", lease_token=admission.lease_token, error="again")
+    with execution_store() as session:
+        assert len(session.scalars(select(TransferExecutionStep)).all()) == 2
+    snapshot = repository.get_snapshot(task_id="task-1")
+    assert len(snapshot.steps) == 1
+    assert snapshot.steps[0].state is TransferStepState.SUCCEEDED
+    assert snapshot.retry_count == 2
+
+
 def test_stable_operation_and_checkpoint_identities_are_canonical():
    """字段顺序不能改变 operation ID 或执行 checkpoint 指纹。"""
    first = build_transfer_operation_id(
--- a/tests/test_transfer_worker_lifecycle.py
+++ b/tests/test_transfer_worker_lifecycle.py
@@ -925,6 +925,45 @@ def test_worker_manual_review_logs_queue_guidance_without_traceback(monkeypatch)
    assert chain._queue.unfinished_tasks == 0


+def test_worker_isolates_exception_from_plugin_replacement_finally(monkeypatch) -> None:
+    """插件替换整个执行方法并在 finally 抛异常时，worker 外层仍必须隔离持久任务。"""
+    chain = _build_chain()
+    task = make_task(1)
+    task.bind_admission_task_id("plugin-task")
+    task.bind_execution_lease(owner_id="worker-owner", lease_token="plugin-lease")
+    chain._owned_leases = {"plugin-task": ("plugin-lease", time.monotonic() + 120)}
+    chain.jobview = MagicMock()
+    chain.jobview.pending_total.return_value = 1
+    chain._progress = MagicMock()
+    chain._active_tasks = chain._processed_num = chain._fail_num = chain._total_num = 0
+    chain._transfer_executions = MagicMock()
+    chain._transfer_executions.isolate_unhandled_failure.return_value = True
+    chain.transfer_execution_repository = chain._transfer_executions
+    chain._TransferChain__fail_transfer_task = MagicMock()
+    stop_event = threading.Event()
+
+    def plugin_replacement(*, task, callback):
+        """重现第三方补丁在 finally 调用已不存在的私有方法。"""
+        del callback
+        try:
+            return True, ""
+        finally:
+            stop_event.set()
+            chain._TransferChain__finish_scrape_batch_task(task)
+
+    chain._TransferChain__handle_transfer = plugin_replacement
+    chain._queue.put(TransferQueue(task=task))
+    monkeypatch.setattr(global_vars, "STOP_EVENT", threading.Event())
+    chain._TransferChain__start_transfer(stop_event)
+
+    isolation = chain._transfer_executions.isolate_unhandled_failure
+    isolation.assert_called_once()
+    assert isolation.call_args.kwargs["task_id"] == "plugin-task"
+    assert "_TransferChain__finish_scrape_batch_task" in isolation.call_args.kwargs["error"]
+    assert chain._queue.unfinished_tasks == 0
+    assert chain._owned_leases == {}
+
+
 def test_heartbeat_refreshes_current_token_and_forgets_lost_lease() -> None:
    """heartbeat 成功应刷新本地期限，CAS 拒绝后必须立即停止本地推进资格。"""
    chain = _build_chain()