diff --git a/app/runtime/extensions/module/contracts.py b/app/runtime/extensions/module/contracts.py index 0afdd2a36..bd030ff69 100644 --- a/app/runtime/extensions/module/contracts.py +++ b/app/runtime/extensions/module/contracts.py @@ -17,6 +17,7 @@ class ModuleResultAggregation(StrEnum): ORDERED_LIST_MERGE = "ordered_list_merge" ORDERED_MAPPING_MERGE = "ordered_mapping_merge" PIPELINE_RELAY = "pipeline_relay" + FAN_OUT = "fan_out" class ModuleResultShape(StrEnum): @@ -118,6 +119,7 @@ _METHOD_CONTRACTS = { "snapshot_storage": ModuleMethodContract(family="storage", input_contract="StorageSnapshotRequest", result_contract="dict[str, dict] | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("storage", "path", "last_snapshot_time", "max_depth", "previous_snapshot")), "transfer": ModuleMethodContract(family="storage", input_contract="TransferRequest", result_contract="TransferInfo | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("fileitem", "meta", "mediainfo", "target_directory", "target_storage", "target_path", "transfer_type", "scrape", "library_type_folder", "library_category_folder", "episodes_info", "source_oper", "target_oper", "preview")), "load_category_config": ModuleMethodContract(family="category", input_contract="CategoryConfigReadRequest", result_contract="CategoryConfig | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY), + "clear_cache": ModuleMethodContract(family="category", input_contract="CacheClearRequest", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, plugin_short_circuit=False), "save_category_config": ModuleMethodContract(family="category", input_contract="CategoryConfigWriteRequest", result_contract="bool | None", result_shape=ModuleResultShape.BOOLEAN, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("config",)), "get_search_page_size": ModuleMethodContract(family="site", input_contract="SiteSearchPageSizeRequest", result_contract="int | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("site", "keyword")), "refresh_userdata": ModuleMethodContract(family="site", input_contract="SiteUserDataRequest", result_contract="SiteUserData | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("site",)), @@ -188,6 +190,7 @@ _METHOD_CONTRACTS = { "music_artist_related": ModuleMethodContract(family="music", input_contract="MusicRelatedRequest", result_contract="list[MusicArtistInfo]", result_shape=ModuleResultShape.LIST, aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE, required_parameters=("media_source", "media_id", "count")), "music_cache_delete": ModuleMethodContract(family="music", input_contract="MusicCacheDeleteRequest", result_contract="dict | None", result_shape=ModuleResultShape.MAPPING, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("cache_key",)), "music_cache_items": ModuleMethodContract(family="music", input_contract="MusicCacheReadRequest", result_contract="list[dict]", result_shape=ModuleResultShape.LIST, aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE), + "music_cache_clear": ModuleMethodContract(family="music", input_contract="MusicCacheClearRequest", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, plugin_short_circuit=False), "music_chart": ModuleMethodContract(family="music", input_contract="MusicChartRequest", result_contract="list[MusicInfo]", result_shape=ModuleResultShape.LIST, aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE, required_parameters=("range_name", "offset", "count", "entity")), "music_discover": ModuleMethodContract(family="music", input_contract="MusicDiscoverRequest", result_contract="list[MusicInfo]", result_shape=ModuleResultShape.LIST, aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE, required_parameters=("media_source", "page", "count", "entity", "mode", "tags", "sort")), "music_fresh_releases": ModuleMethodContract(family="music", input_contract="MusicFreshReleasesRequest", result_contract="list[MusicInfo]", result_shape=ModuleResultShape.LIST, aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE, required_parameters=("days", "sort", "past", "future", "offset", "count")), @@ -202,8 +205,8 @@ _METHOD_CONTRACTS = { "send_direct_message": ModuleMethodContract(family="messaging", input_contract="DirectMessageSendRequest", result_contract="MessageResponse | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("message",)), "send_message": ModuleMethodContract(family="messaging", input_contract="MessageSendRequest", result_contract="Message | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY), "finalize_message": ModuleMethodContract(family="messaging", input_contract="MessageFinalizeRequest", result_contract="Message | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("response",)), - "register_commands": ModuleMethodContract(family="messaging", input_contract="CommandRegistrationRequest", result_contract="None", required_parameters=("commands",)), - "scheduler_job": ModuleMethodContract(family="scheduling", input_contract="SchedulerJobRequest", result_contract="None"), + "register_commands": ModuleMethodContract(family="messaging", input_contract="CommandRegistrationRequest", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, required_parameters=("commands",), plugin_short_circuit=False), + "scheduler_job": ModuleMethodContract(family="scheduling", input_contract="SchedulerJobRequest", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, plugin_short_circuit=False), "webhook_parser": ModuleMethodContract(family="integration", input_contract="WebhookRequest", result_contract="WebhookEventInfo | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("body", "form", "args")), "download_discord_file_bytes": ModuleMethodContract(family="messaging", input_contract="MessageFileDownloadRequest", result_contract="bytes | None", result_shape=ModuleResultShape.BYTES, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("file_ref", "source")), "download_feishu_file_bytes": ModuleMethodContract(family="messaging", input_contract="MessageFileDownloadRequest", result_contract="bytes | None", result_shape=ModuleResultShape.BYTES, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("file_ref", "source")), @@ -225,13 +228,16 @@ _METHOD_CONTRACTS = { "torrent_files": ModuleMethodContract(family="downloader", input_contract="TorrentFilesRequest", result_contract="DownloaderFileCollection | None", required_parameters=("tid", "downloader")), "get_torrent_trackers": ModuleMethodContract(family="downloader", input_contract="TorrentTrackersRequest", result_contract="dict[str, list[str]] | None", result_shape=ModuleResultShape.MAPPING, aggregation=ModuleResultAggregation.ORDERED_MAPPING_MERGE, required_parameters=("hash_string", "downloader")), "download": ModuleMethodContract(family="downloader", input_contract="DownloadTaskRequest", result_contract="DownloadTaskResult | None", aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("content", "download_dir", "cookie", "episodes", "category", "label", "downloader")), + "download_added": ModuleMethodContract(family="downloader", input_contract="DownloadAddedHook", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, required_parameters=("context", "torrent_content", "download_dir"), plugin_short_circuit=False), "remove_torrents": ModuleMethodContract(family="downloader", input_contract="TorrentRemoveRequest", result_contract="bool | None", result_shape=ModuleResultShape.BOOLEAN, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("hashs", "delete_file", "downloader")), "set_torrents_tag": ModuleMethodContract(family="downloader", input_contract="TorrentTagRequest", result_contract="bool | None", result_shape=ModuleResultShape.BOOLEAN, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("hashs", "tags", "downloader")), "start_torrents": ModuleMethodContract(family="downloader", input_contract="TorrentControlRequest", result_contract="bool | None", result_shape=ModuleResultShape.BOOLEAN, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("hashs", "downloader")), "stop_torrents": ModuleMethodContract(family="downloader", input_contract="TorrentControlRequest", result_contract="bool | None", result_shape=ModuleResultShape.BOOLEAN, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("hashs", "downloader")), "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")), + "transfer_completed": ModuleMethodContract(family="downloader", input_contract="TransferCompletedHook", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, required_parameters=("hashs", "downloader"), plugin_short_circuit=False), "tmdb_cache_items": ModuleMethodContract(family="tmdb", input_contract="TmdbCacheListRequest", result_contract="list[dict[str, Any]]", result_shape=ModuleResultShape.LIST, aggregation=ModuleResultAggregation.ORDERED_LIST_MERGE), "tmdb_cache_delete": ModuleMethodContract(family="tmdb", input_contract="TmdbCacheDeleteRequest", result_contract="dict[str, Any] | None", result_shape=ModuleResultShape.MAPPING, aggregation=ModuleResultAggregation.FIRST_NON_EMPTY, required_parameters=("cache_key",)), + "tmdb_cache_clear": ModuleMethodContract(family="tmdb", input_contract="TmdbCacheClearRequest", result_contract="None", aggregation=ModuleResultAggregation.FAN_OUT, plugin_short_circuit=False), } # 同一能力的同步/异步入口共享不可变契约对象，避免参数和聚合语义各自漂移。 diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py index 140f08ce1..704b6e88d 100644 --- a/app/runtime/extensions/module/dispatcher.py +++ b/app/runtime/extensions/module/dispatcher.py @@ -85,7 +85,11 @@ class ModuleInvocationDispatcher: "module.provider.duration", method=method, provider_type="plugin" ): result = self.execute_plugin_modules(method, None, *args, **kwargs) - if not self.is_valid_empty(result) and not isinstance(result, list): + if ( + contract.plugin_short_circuit + and not self.is_valid_empty(result) + and not isinstance(result, list) + ): return result with observe_duration( "module.provider.duration", method=method, provider_type="system" @@ -105,7 +109,11 @@ class ModuleInvocationDispatcher: *args, **kwargs, ) - if not self.is_valid_empty(result) and not isinstance(result, list): + if ( + contract.plugin_short_circuit + and not self.is_valid_empty(result) + and not isinstance(result, list) + ): return result with observe_duration( "module.provider.duration", method=method, provider_type="system" @@ -158,6 +166,7 @@ class ModuleInvocationDispatcher: result, provider_result, call_mode, + aggregation, ) except RateLimitExceededException as err: self._rate_limit_handler( @@ -219,6 +228,7 @@ class ModuleInvocationDispatcher: result, provider_result, call_mode, + aggregation, ) except RateLimitExceededException as err: self._rate_limit_handler( @@ -281,6 +291,7 @@ class ModuleInvocationDispatcher: result, provider_result, call_mode, + aggregation, ) except RateLimitExceededException as err: self._rate_limit_handler( @@ -343,6 +354,7 @@ class ModuleInvocationDispatcher: result, provider_result, call_mode, + aggregation, ) except RateLimitExceededException as err: self._rate_limit_handler( @@ -373,6 +385,8 @@ class ModuleInvocationDispatcher: allow_relay: bool, ) -> _ProviderCallMode: """按契约选择下一 provider 的调用方式，并冻结 legacy 接力语义。""" + if aggregation is ModuleResultAggregation.FAN_OUT: + return _ProviderCallMode.ORIGINAL if cls.is_valid_empty(result): return _ProviderCallMode.ORIGINAL if aggregation is ModuleResultAggregation.FIRST_NON_EMPTY: @@ -404,8 +418,11 @@ class ModuleInvocationDispatcher: result: Any, provider_result: Any, call_mode: _ProviderCallMode, + aggregation: ModuleResultAggregation, ) -> Any: """合并单个 provider 结果，接力调用则用新结果替换旧结果。""" + if aggregation is ModuleResultAggregation.FAN_OUT: + return result if call_mode is _ProviderCallMode.RELAY: return provider_result if isinstance(result, list) and isinstance(provider_result, list): diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index b91177f2a..c6b78cb82 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -3762,15 +3762,15 @@ "version": 1 }, "clear_cache": { - "aggregation": "legacy", + "aggregation": "fan_out", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "category", - "input_contract": "CategoryKeywordArguments", - "plugin_short_circuit": true, + "input_contract": "CacheClearRequest", + "plugin_short_circuit": false, "public_to_plugins": true, "required_parameters": [], - "result_contract": "CategoryProviderResult", + "result_contract": "None", "result_shape": "any", "supports_async": true, "supports_sync": true, @@ -4011,15 +4011,19 @@ "version": 1 }, "download_added": { - "aggregation": "legacy", + "aggregation": "fan_out", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "downloader", - "input_contract": "DownloaderKeywordArguments", - "plugin_short_circuit": true, + "input_contract": "DownloadAddedHook", + "plugin_short_circuit": false, "public_to_plugins": true, - "required_parameters": [], - "result_contract": "DownloaderProviderResult", + "required_parameters": [ + "context", + "download_dir", + "torrent_content" + ], + "result_contract": "None", "result_shape": "any", "supports_async": true, "supports_sync": true, @@ -5160,15 +5164,15 @@ "version": 1 }, "music_cache_clear": { - "aggregation": "legacy", + "aggregation": "fan_out", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "music", - "input_contract": "MusicKeywordArguments", - "plugin_short_circuit": true, + "input_contract": "MusicCacheClearRequest", + "plugin_short_circuit": false, "public_to_plugins": true, "required_parameters": [], - "result_contract": "MusicProviderResult", + "result_contract": "None", "result_shape": "any", "supports_async": true, "supports_sync": true, @@ -5420,12 +5424,12 @@ "version": 1 }, "register_commands": { - "aggregation": "legacy", + "aggregation": "fan_out", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "messaging", "input_contract": "CommandRegistrationRequest", - "plugin_short_circuit": true, + "plugin_short_circuit": false, "public_to_plugins": true, "required_parameters": [ "commands" @@ -5495,12 +5499,12 @@ "version": 1 }, "scheduler_job": { - "aggregation": "legacy", + "aggregation": "fan_out", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "scheduling", "input_contract": "SchedulerJobRequest", - "plugin_short_circuit": true, + "plugin_short_circuit": false, "public_to_plugins": true, "required_parameters": [], "result_contract": "None", @@ -5798,15 +5802,15 @@ "version": 1 }, "tmdb_cache_clear": { - "aggregation": "legacy", + "aggregation": "fan_out", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "tmdb", - "input_contract": "TmdbKeywordArguments", - "plugin_short_circuit": true, + "input_contract": "TmdbCacheClearRequest", + "plugin_short_circuit": false, "public_to_plugins": true, "required_parameters": [], - "result_contract": "TmdbProviderResult", + "result_contract": "None", "result_shape": "any", "supports_async": true, "supports_sync": true, @@ -6184,15 +6188,18 @@ "version": 1 }, "transfer_completed": { - "aggregation": "legacy", + "aggregation": "fan_out", "error_policy": "isolate_provider", "execution": "sync_or_async", "family": "downloader", - "input_contract": "DownloaderKeywordArguments", - "plugin_short_circuit": true, + "input_contract": "TransferCompletedHook", + "plugin_short_circuit": false, "public_to_plugins": true, - "required_parameters": [], - "result_contract": "DownloaderProviderResult", + "required_parameters": [ + "downloader", + "hashs" + ], + "result_contract": "None", "result_shape": "any", "supports_async": true, "supports_sync": true, diff --git a/tests/test_module_invocation_dispatcher.py b/tests/test_module_invocation_dispatcher.py index d298f8512..45cf6a3b3 100644 --- a/tests/test_module_invocation_dispatcher.py +++ b/tests/test_module_invocation_dispatcher.py @@ -94,6 +94,56 @@ def test_plugin_scalar_short_circuits_system_modules() -> None: system_call.assert_not_called() +def test_fan_out_contract_runs_every_provider_and_ignores_results() -> None: + """副作用广播应执行全部插件和宿主 provider，并稳定返回 None。""" + calls = [] + + def record(name: str, result): + """生成记录调用顺序并返回测试哨兵的 provider。""" + return lambda: calls.append(name) or result + + system_20 = _Module("系统二", 20, record("system-20", "ignored-system")) + system_10 = _Module("系统一", 10, record("system-10", None)) + setattr(system_20, "clear_cache", system_20.execute) + setattr(system_10, "clear_cache", system_10.execute) + dispatcher, _, _, _ = _dispatcher( + plugins={ + ("P1", "插件一"): {"clear_cache": record("plugin-1", "ignored-plugin")}, + ("P2", "插件二"): {"clear_cache": record("plugin-2", None)}, + }, + modules=[system_20, system_10], + ) + + assert dispatcher.dispatch("clear_cache") is None + assert calls == ["plugin-1", "plugin-2", "system-10", "system-20"] + + +@pytest.mark.asyncio +async def test_async_fan_out_contract_matches_sync_execution() -> None: + """异步广播也应忽略返回值并执行全部同步或异步 provider。""" + calls = [] + + async def plugin_call(): + """记录异步插件调用并返回应被忽略的哨兵。""" + calls.append("plugin") + return "ignored-plugin" + + def system_call(): + """记录同步宿主调用并返回应被忽略的哨兵。""" + calls.append("system") + return "ignored-system" + + module = _Module("系统", 10, system_call) + setattr(module, "clear_cache", module.execute) + dispatcher, _, _, _ = _dispatcher( + plugins={("P1", "插件一"): {"clear_cache": plugin_call}}, + modules=[module], + ) + + assert await dispatcher.async_dispatch("clear_cache") is None + assert calls == ["plugin", "system"] + + def test_list_results_merge_in_plugin_then_priority_order() -> None: """列表结果应先按插件顺序合并，再按宿主优先级继续合并。""" calls = [] diff --git a/tests/test_module_method_contracts.py b/tests/test_module_method_contracts.py index 4dc7522ce..24ba7a181 100644 --- a/tests/test_module_method_contracts.py +++ b/tests/test_module_method_contracts.py @@ -34,7 +34,7 @@ def test_all_scanned_host_module_methods_have_explicit_v2_contracts() -> None: assert isinstance(contract.aggregation, ModuleResultAggregation) assert contract.input_contract != "legacy_args" assert contract.result_contract != "Any" - assert contract.plugin_short_circuit is True + assert isinstance(contract.plugin_short_circuit, bool) def test_high_frequency_capability_families_are_explicit() -> None: @@ -298,6 +298,25 @@ def test_torrent_filter_contract_preserves_original_argument_list_merge() -> Non ) +def test_side_effect_hooks_use_non_short_circuiting_fan_out_contracts() -> None: + """副作用钩子必须执行全部 provider，不能被任意返回值提前截断。""" + methods = { + "clear_cache", + "download_added", + "music_cache_clear", + "register_commands", + "scheduler_job", + "tmdb_cache_clear", + "transfer_completed", + } + + for method in methods: + contract = get_module_method_contract(method) + assert contract.aggregation is ModuleResultAggregation.FAN_OUT + assert contract.result_contract == "None" + assert contract.plugin_short_circuit is False + + def test_heterogeneous_torrent_files_result_remains_legacy_compatible() -> None: """下载器文件集合尚未归一前不得声明虚假的列表聚合语义。""" contract = get_module_method_contract("torrent_files")