diff --git a/app/api/endpoints/douban.py b/app/api/endpoints/douban.py index bc334a3bf..0bfc2cdb8 100644 --- 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: diff --git a/app/api/endpoints/mediaserver.py b/app/api/endpoints/mediaserver.py index e2c99f5c6..231106dc8 100644 --- 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 ) diff --git a/app/api/endpoints/tmdb.py b/app/api/endpoints/tmdb.py index 412bab36b..88ff98dea 100644 --- 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: diff --git a/app/api/endpoints/transfer.py b/app/api/endpoints/transfer.py index f7b5dd79e..b423d6b1a 100644 --- 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]: """为未显式指定实体的旧客户端按源项类型补全音乐命名空间。""" diff --git a/app/application/transfer/execution.py b/app/application/transfer/execution.py index 729c9c2f3..e97effedd 100644 --- 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, *, diff --git a/app/chain/transfer/execution.py b/app/chain/transfer/execution.py index bcfd7553c..9d84e8ce3 100644 --- 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} 已转入人工复核，未自动重放；" diff --git a/app/db/adapters/transfer/execution.py b/app/db/adapters/transfer/execution.py index 546c5e61e..1d608ea50 100644 --- 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( "这条整理步骤无法确认是否已经执行，请人工确认文件状态后再继续；" diff --git a/app/db/models/transferpending.py b/app/db/models/transferpending.py index 56993634d..e1458df92 100644 --- 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, diff --git a/app/db/oper/transferpending.py b/app/db/oper/transferpending.py index b03e17c97..7802c97be 100644 --- 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], diff --git a/docs/mcp-api.md b/docs/mcp-api.md index c56d9e2a7..aa2296707 100644 --- 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 的存储访问路径是两个地址空间。下载器配置的 diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 1f2b0bff3..ae91fb3df 100644 --- 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. diff --git a/tests/test_api_media_types.py b/tests/test_api_media_types.py new file mode 100644 index 000000000..0b635710e --- /dev/null +++ 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() diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index 27ebaaecd..1c4201d24 100644 --- 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", diff --git a/tests/test_transfer_execution_persistence.py b/tests/test_transfer_execution_persistence.py index 07046614d..aae6a0c13 100644 --- 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( diff --git a/tests/test_transfer_worker_lifecycle.py b/tests/test_transfer_worker_lifecycle.py index 0e7edd365..9d4f921f6 100644 --- 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()