diff --git a/app/runtime/extensions/module/contracts.py b/app/runtime/extensions/module/contracts.py index 07c2af3b8..f655d2904 100644 --- a/app/runtime/extensions/module/contracts.py +++ b/app/runtime/extensions/module/contracts.py @@ -16,6 +16,7 @@ class ModuleResultAggregation(StrEnum): FIRST_NON_EMPTY = "first_non_empty" ORDERED_LIST_MERGE = "ordered_list_merge" ORDERED_MAPPING_MERGE = "ordered_mapping_merge" + PIPELINE_RELAY = "pipeline_relay" class ModuleResultShape(StrEnum): @@ -80,10 +81,11 @@ _METHOD_CONTRACTS = { ), "search_medias": ModuleMethodContract( family="media-recognition", input_contract="MediaSearchRequest", - result_contract="list[MediaInfo]", aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE, + result_contract="list[MediaInfo]", result_shape=ModuleResultShape.LIST, + aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE, required_parameters=("meta", "media_source"), ), - "obtain_images": ModuleMethodContract(family="media-recognition", input_contract="MediaInfo", result_contract="MediaInfo | None", required_parameters=("mediainfo",)), + "obtain_images": ModuleMethodContract(family="media-recognition", input_contract="MediaInfo", result_contract="MediaInfo | None", aggregation=ModuleResultAggregation.PIPELINE_RELAY, required_parameters=("mediainfo",)), "media_category": ModuleMethodContract(family="media-recognition", input_contract="MediaCategoryRequest", result_contract="CategoryConfig | None"), "mediaserver_items": ModuleMethodContract(family="media-server", input_contract="MediaServerItemsRequest", result_contract="list[MediaServerItem]", aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE, required_parameters=("server", "library_id", "start_index", "limit")), "mediaserver_iteminfo": ModuleMethodContract(family="media-server", input_contract="MediaServerItemRequest", result_contract="MediaServerItem | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("server", "item_id")), @@ -128,6 +130,13 @@ _METHOD_CONTRACTS = { "update_torrent": ModuleMethodContract(family="downloader", input_contract="TorrentUpdateRequest", result_contract="dict[str, bool] | None", result_shape=ModuleResultShape.MAPPING, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("hash_string", "downloader", "download_limit", "upload_limit", "tracker_list", "save_path", "category", "ratio_limit", "seeding_time_limit")), } +# 同一能力的同步/异步入口共享不可变契约对象，避免参数和聚合语义各自漂移。 +_METHOD_CONTRACTS.update({ + "async_recognize_media": _METHOD_CONTRACTS["recognize_media"], + "async_search_medias": _METHOD_CONTRACTS["search_medias"], + "async_obtain_images": _METHOD_CONTRACTS["obtain_images"], +}) + _PREFIX_CONTRACTS = ( ("async_tmdb_", ModuleMethodContract(family="tmdb")), ("tmdb_", ModuleMethodContract(family="tmdb")), diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py index 140560dbb..140f08ce1 100644 --- a/app/runtime/extensions/module/dispatcher.py +++ b/app/runtime/extensions/module/dispatcher.py @@ -389,10 +389,14 @@ class ModuleInvocationDispatcher: if isinstance(result, dict) else _ProviderCallMode.STOP ) - if allow_relay and ObjectUtils.check_signature(func, result): - return _ProviderCallMode.RELAY - if isinstance(result, list): - return _ProviderCallMode.ORIGINAL + if aggregation in { + ModuleResultAggregation.LEGACY, + ModuleResultAggregation.PIPELINE_RELAY, + }: + if allow_relay and ObjectUtils.check_signature(func, result): + return _ProviderCallMode.RELAY + if isinstance(result, list): + return _ProviderCallMode.ORIGINAL return _ProviderCallMode.STOP @staticmethod diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 0474af722..faa3d18da 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -91,7 +91,7 @@ `shield` 不再让网络请求逃逸生命周期预算，仓库级并发合并、缓存键和 V1/V2/V3 返回兼容保持不变。 请求作用域的结构化并发不进入全局登记器：传统 WebAgent SSE 的 collection 子任务改由生成器 `finally` 取消并等待清理，断线和 ASGI 取消均不会留下请求级 task。 -2. **动态模块契约仍以 legacy 聚合语义为主。** 当前登记 `212` 个模块方法，其中 `172` 个仍使用 `legacy` aggregation，`33` 个使用 `first_non_empty`、`6` 个使用 `ordered_list_merge`、`1` 个使用 `ordered_mapping_merge`。`app/runtime/extensions/module/contracts.py` 已能登记 family、输入/结果标签和基础签名诊断，调度器也已按这 40 个显式聚合声明执行首个非空、有序列表或有序映射合并；但 `170` 个方法没有 required parameters，其余方法仍主要依赖运行时反射、返回值形状和旧短路规则。未知第三方方法保留 legacy fallback 是兼容要求，不应删除；宿主高频能力则应逐族补齐可执行的输入校验、结果校验、超时和错误语义。 +2. **动态模块契约仍以 legacy 聚合语义为主。** 当前登记 `212` 个模块方法，其中 `168` 个仍使用 `legacy` aggregation，`34` 个使用 `first_non_empty`、`7` 个使用 `ordered_list_merge`、`1` 个使用 `ordered_mapping_merge`、`2` 个使用 `pipeline_relay`。`app/runtime/extensions/module/contracts.py` 已能登记 family、输入/结果标签和基础签名诊断，调度器也已按这 44 个显式聚合声明执行首个非空、有序集合合并或接力管道；但 `167` 个方法没有 required parameters，其余方法仍主要依赖运行时反射、返回值形状和旧短路规则。未知第三方方法保留 legacy fallback 是兼容要求，不应删除；宿主高频能力则应逐族补齐可执行的输入校验、结果校验、超时和错误语义。 3. **Model/Base 的数据库装饰器和隐式会话 ABI 已全部清零。** 查询、写事务和 `legacy_*` 装饰器均为 `0`；所有 Model `db` 参数要求显式 Session，Base CRUD 仅在调用方事务内查询或 stage。可无会话构造的入口统一留在 Oper，经组合根事务执行器运行；插件 SDK 不再导出宿主 Model。后续重点转为减少 ORM 对象跨层流转，并保持 Model 隐式事务零回退。 Oper 内部的执行入口也已统一：最后一处 `AgentTaskOper` 直接 transaction runner 调用已迁入 @@ -804,6 +804,10 @@ ModuleMethodSpec( - qBittorrent、Transmission、rTorrent 共享的 `download`、删除、启停、标签与更新 6 个目标选择动作已冻结 一致参数和首个非空结果语义；bool/dict 结果启用基础形状诊断，tuple 下载结果保持业务合同标签而不强制 Python 形状。广播型 `download_added` / `transfer_completed` 和管道型 `filter_torrents` 继续保留 legacy。 +- 识别与搜索的 sync/async 方法现在复用同一个不可变契约对象，`async_recognize_media` 与 + `async_search_medias` 不再落入不同 family/aggregation；同步、异步 `obtain_images` 也统一登记为显式 + `pipeline_relay`，按宿主优先级把同一 `MediaInfo` 交给后续图片 provider。未知插件方法和 legacy 接力 + 仍使用原算法，插件先返回非空对象时继续优先短路宿主。 #### ARCH-241：Event Contract Registry diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index ef3434743..7de4dc104 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -2972,15 +2972,17 @@ "version": 1 }, "async_obtain_images": { - "aggregation": "legacy", + "aggregation": "pipeline_relay", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "media-recognition", - "input_contract": "MediaRecognitionKeywordArguments", + "input_contract": "MediaInfo", "plugin_short_circuit": true, "public_to_plugins": true, - "required_parameters": [], - "result_contract": "MediaRecognitionProviderResult", + "required_parameters": [ + "mediainfo" + ], + "result_contract": "MediaInfo | None", "result_shape": "any", "supports_async": true, "supports_sync": true, @@ -2988,15 +2990,22 @@ "version": 1 }, "async_recognize_media": { - "aggregation": "legacy", + "aggregation": "first_non_empty", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "media-recognition", - "input_contract": "MediaRecognitionKeywordArguments", + "input_contract": "MediaRecognitionRequest", "plugin_short_circuit": true, "public_to_plugins": true, - "required_parameters": [], - "result_contract": "MediaRecognitionProviderResult", + "required_parameters": [ + "cache", + "episode_group", + "media_id", + "media_source", + "meta", + "mtype" + ], + "result_contract": "MediaInfo | None", "result_shape": "any", "supports_async": true, "supports_sync": true, @@ -3036,16 +3045,19 @@ "version": 1 }, "async_search_medias": { - "aggregation": "legacy", + "aggregation": "ordered_list_merge", "error_policy": "isolate_provider", "execution": "sync_or_async", - "family": "media-discovery", - "input_contract": "MediaDiscoveryKeywordArguments", + "family": "media-recognition", + "input_contract": "MediaSearchRequest", "plugin_short_circuit": true, "public_to_plugins": true, - "required_parameters": [], - "result_contract": "MediaDiscoveryProviderResult", - "result_shape": "any", + "required_parameters": [ + "media_source", + "meta" + ], + "result_contract": "list[MediaInfo]", + "result_shape": "list", "supports_async": true, "supports_sync": true, "timeout_policy": "caller_budget", @@ -4899,7 +4911,7 @@ "version": 1 }, "obtain_images": { - "aggregation": "legacy", + "aggregation": "pipeline_relay", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "media-recognition", @@ -5121,7 +5133,7 @@ "meta" ], "result_contract": "list[MediaInfo]", - "result_shape": "any", + "result_shape": "list", "supports_async": true, "supports_sync": true, "timeout_policy": "caller_budget", diff --git a/tests/test_module_invocation_dispatcher.py b/tests/test_module_invocation_dispatcher.py index 418c2520a..d298f8512 100644 --- a/tests/test_module_invocation_dispatcher.py +++ b/tests/test_module_invocation_dispatcher.py @@ -167,6 +167,43 @@ def test_system_signature_relay_passes_previous_result() -> None: assert dispatcher.dispatch("execute") == {"value": 2} +def test_explicit_pipeline_contract_relays_previous_result() -> None: + """图片补全契约应按优先级把上一 provider 结果交给下一 provider。""" + class ImageModule: + """在统一媒体对象上记录当前图片 provider。""" + + def __init__(self, name: str, priority: int) -> None: + """保存 provider 名称和优先级。""" + self._name = name + self._priority = priority + + def get_name(self) -> str: + """返回测试模块名。""" + return self._name + + def get_priority(self) -> int: + """返回测试优先级。""" + return self._priority + + def obtain_images(self, mediainfo: dict) -> dict: + """追加当前 provider 名称并返回同一媒体结果。""" + return { + **mediainfo, + "providers": [*mediainfo.get("providers", []), self._name], + } + + dispatcher, _, _, _ = _dispatcher( + modules=[ + ImageModule("fanart", 20), + ImageModule("tmdb", 10), + ] + ) + + assert dispatcher.dispatch("obtain_images", mediainfo={}) == { + "providers": ["tmdb", "fanart"] + } + + def test_first_non_empty_contract_stops_legacy_signature_relay() -> None: """显式首个非空契约不得再把结果交给后续宿主 provider 改写。""" class FirstModule: diff --git a/tests/test_module_method_contracts.py b/tests/test_module_method_contracts.py index 30ed053b7..267b5a57c 100644 --- a/tests/test_module_method_contracts.py +++ b/tests/test_module_method_contracts.py @@ -214,6 +214,23 @@ def test_downloader_action_contracts_freeze_shared_provider_signatures() -> None assert contract.required_parameters == parameters +def test_sync_and_async_media_capabilities_share_contracts() -> None: + """同一识别能力的同步与异步入口必须复用完全相同的契约。""" + for sync_method, async_method in ( + ("recognize_media", "async_recognize_media"), + ("search_medias", "async_search_medias"), + ("obtain_images", "async_obtain_images"), + ): + assert get_module_method_contract(sync_method) is get_module_method_contract( + async_method + ) + + assert ( + get_module_method_contract("obtain_images").aggregation + is ModuleResultAggregation.PIPELINE_RELAY + ) + + def test_attachment_result_diagnostics_distinguish_bytes_and_strings() -> None: """附件契约应区分二进制内容和可展示字符串，偏差仍仅供诊断。""" assert diagnose_module_result("download_qq_file_bytes", b"content") == ()