From bec64887d63f135a2f84c827f30601a649984cc7 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 24 Aug 2026 00:57:19 +0800 Subject: [PATCH] refactor: enforce module aggregation contracts --- app/runtime/extensions/module/dispatcher.py | 157 +++++++++++++----- .../backend-architecture-next-stage.md | 6 +- tests/test_module_invocation_dispatcher.py | 113 +++++++++++++ 3 files changed, 233 insertions(+), 43 deletions(-) diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py index cf5463dcc..0718c10de 100644 --- a/app/runtime/extensions/module/dispatcher.py +++ b/app/runtime/extensions/module/dispatcher.py @@ -4,6 +4,7 @@ from __future__ import annotations import inspect from collections.abc import Callable, Mapping +from enum import StrEnum from typing import Any, Protocol, cast from app.foundation.reflection import ObjectUtils @@ -11,6 +12,7 @@ from app.runtime.execution import run_in_threadpool_to_completion from app.runtime.log import logger from app.runtime.observability import observe_duration, record_metric from app.runtime.extensions.module.contracts import ( + ModuleResultAggregation, diagnose_module_callable, diagnose_module_result, get_module_method_contract, @@ -39,6 +41,14 @@ ModuleErrorHandler = Callable[..., None] AsyncFunctionRunner = Callable[..., Any] +class _ProviderCallMode(StrEnum): + """描述当前 provider 应采用的兼容调用方式。""" + + ORIGINAL = "original" + RELAY = "relay" + STOP = "stop" + + class ModuleInvocationDispatcher: """按既有聚合、短路和异常规则执行插件与宿主模块。""" @@ -115,6 +125,7 @@ class ModuleInvocationDispatcher: **kwargs: Any, ) -> Any: """同步执行插件方法，保留插件顺序、短路和列表合并语义。""" + aggregation = get_module_method_contract(method).aggregation for plugin, module_dict in self._plugin_catalog.get_plugin_modules().items(): plugin_id, plugin_name = plugin try: @@ -133,16 +144,21 @@ class ModuleInvocationDispatcher: ) self._diagnose_callable(method, func, f"插件 {plugin_id}") logger.info("请求插件 %s 执行：%s ...", plugin_name, method) - if self.is_valid_empty(result): - result = func(*args, **kwargs) - self._diagnose_result(method, result, "plugin") - elif isinstance(result, list): - temp = func(*args, **kwargs) - self._diagnose_result(method, temp, "plugin") - if isinstance(temp, list): - result.extend(temp) - else: + call_mode = self._provider_call_mode( + aggregation, + result, + func, + allow_relay=False, + ) + if call_mode is _ProviderCallMode.STOP: break + provider_result = func(*args, **kwargs) + self._diagnose_result(method, provider_result, "plugin") + result = self._aggregate_provider_result( + result, + provider_result, + call_mode, + ) except RateLimitExceededException as err: self._rate_limit_handler( err, @@ -170,6 +186,7 @@ class ModuleInvocationDispatcher: **kwargs: Any, ) -> Any: """异步执行插件方法，并把同步函数移入线程池。""" + aggregation = get_module_method_contract(method).aggregation for plugin, module_dict in self._plugin_catalog.get_plugin_modules().items(): plugin_id, plugin_name = plugin try: @@ -188,16 +205,21 @@ class ModuleInvocationDispatcher: ) self._diagnose_callable(method, func, f"插件 {plugin_id}") logger.info("请求插件 %s 执行：%s ...", plugin_name, method) - if self.is_valid_empty(result): - result = await self._async_call(func, *args, **kwargs) - self._diagnose_result(method, result, "plugin") - elif isinstance(result, list): - temp = await self._async_call(func, *args, **kwargs) - self._diagnose_result(method, temp, "plugin") - if isinstance(temp, list): - result.extend(temp) - else: + call_mode = self._provider_call_mode( + aggregation, + result, + func, + allow_relay=False, + ) + if call_mode is _ProviderCallMode.STOP: break + provider_result = await self._async_call(func, *args, **kwargs) + self._diagnose_result(method, provider_result, "plugin") + result = self._aggregate_provider_result( + result, + provider_result, + call_mode, + ) except RateLimitExceededException as err: self._rate_limit_handler( err, @@ -226,6 +248,7 @@ class ModuleInvocationDispatcher: ) -> Any: """同步执行按优先级排序的宿主模块，并支持签名接力。""" logger.debug("请求系统模块执行：%s ...", method) + aggregation = get_module_method_contract(method).aggregation modules = sorted( self._module_catalog.get_running_modules(method), key=lambda module: module.get_priority(), @@ -241,19 +264,24 @@ class ModuleInvocationDispatcher: abi_source="host_module", ) self._diagnose_callable(method, func, f"宿主模块 {module_id}") - if self.is_valid_empty(result): - result = func(*args, **kwargs) - self._diagnose_result(method, result, "system") - elif ObjectUtils.check_signature(func, result): - result = func(result) - self._diagnose_result(method, result, "system") - elif isinstance(result, list): - temp = func(*args, **kwargs) - self._diagnose_result(method, temp, "system") - if isinstance(temp, list): - result.extend(temp) - else: + call_mode = self._provider_call_mode( + aggregation, + result, + func, + allow_relay=True, + ) + if call_mode is _ProviderCallMode.STOP: break + if call_mode is _ProviderCallMode.RELAY: + provider_result = func(result) + else: + provider_result = func(*args, **kwargs) + self._diagnose_result(method, provider_result, "system") + result = self._aggregate_provider_result( + result, + provider_result, + call_mode, + ) except RateLimitExceededException as err: self._rate_limit_handler( err, @@ -282,6 +310,7 @@ class ModuleInvocationDispatcher: ) -> Any: """异步执行宿主模块，并保持同步路径的签名接力与聚合顺序。""" logger.debug("请求系统模块执行：%s ...", method) + aggregation = get_module_method_contract(method).aggregation modules = sorted( self._module_catalog.get_running_modules(method), key=lambda module: module.get_priority(), @@ -297,19 +326,24 @@ class ModuleInvocationDispatcher: abi_source="host_module", ) self._diagnose_callable(method, func, f"宿主模块 {module_id}") - if self.is_valid_empty(result): - result = await self._async_call(func, *args, **kwargs) - self._diagnose_result(method, result, "system") - elif ObjectUtils.check_signature(func, result): - result = await self._async_call(func, result) - self._diagnose_result(method, result, "system") - elif isinstance(result, list): - temp = await self._async_call(func, *args, **kwargs) - self._diagnose_result(method, temp, "system") - if isinstance(temp, list): - result.extend(temp) - else: + call_mode = self._provider_call_mode( + aggregation, + result, + func, + allow_relay=True, + ) + if call_mode is _ProviderCallMode.STOP: break + if call_mode is _ProviderCallMode.RELAY: + provider_result = await self._async_call(func, result) + else: + provider_result = await self._async_call(func, *args, **kwargs) + self._diagnose_result(method, provider_result, "system") + result = self._aggregate_provider_result( + result, + provider_result, + call_mode, + ) except RateLimitExceededException as err: self._rate_limit_handler( err, @@ -329,6 +363,45 @@ class ModuleInvocationDispatcher: ) return result + @classmethod + def _provider_call_mode( + cls, + aggregation: ModuleResultAggregation, + result: Any, + func: Callable[..., Any], + *, + allow_relay: bool, + ) -> _ProviderCallMode: + """按契约选择下一 provider 的调用方式，并冻结 legacy 接力语义。""" + if cls.is_valid_empty(result): + return _ProviderCallMode.ORIGINAL + if aggregation is ModuleResultAggregation.FIRST_NON_EMPTY: + return _ProviderCallMode.STOP + if aggregation is ModuleResultAggregation.ORDERED_LIST_MERGE: + return ( + _ProviderCallMode.ORIGINAL + if isinstance(result, list) + else _ProviderCallMode.STOP + ) + if allow_relay and ObjectUtils.check_signature(func, result): + return _ProviderCallMode.RELAY + if isinstance(result, list): + return _ProviderCallMode.ORIGINAL + return _ProviderCallMode.STOP + + @staticmethod + def _aggregate_provider_result( + result: Any, + provider_result: Any, + call_mode: _ProviderCallMode, + ) -> Any: + """合并单个 provider 结果，接力调用则用新结果替换旧结果。""" + if call_mode is _ProviderCallMode.RELAY or not isinstance(result, list): + return provider_result + if isinstance(provider_result, list): + result.extend(provider_result) + return result + @staticmethod def _record_timeout(method: str, provider_type: str, error: Exception) -> None: """仅把真实超时归入低基数模块超时指标。""" diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index d6664d3d8..782f54d3d 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` 个模块方法，其中 `194` 个仍使用 `legacy` aggregation，只有 `14` 个 `first_non_empty`、`4` 个 `ordered_list_merge`。`app/runtime/extensions/module/contracts.py:422-455` 已能登记 family、输入/结果标签和基础签名诊断，但 `193` 个方法没有 required parameters，调度器 `app/runtime/extensions/module/dispatcher.py:109-260` 仍主要依赖运行时反射、返回值形状和短路规则。未知第三方方法保留 legacy fallback 是兼容要求，不应删除；宿主高频能力则应逐族补齐可执行的输入校验、结果校验、超时和错误语义。 +2. **动态模块契约仍以 legacy 聚合语义为主。** 当前登记 `212` 个模块方法，其中 `194` 个仍使用 `legacy` aggregation，只有 `14` 个 `first_non_empty`、`4` 个 `ordered_list_merge`。`app/runtime/extensions/module/contracts.py` 已能登记 family、输入/结果标签和基础签名诊断，调度器也已按这 18 个显式聚合声明执行首个非空或有序列表合并；但 `193` 个方法没有 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 调用已迁入 @@ -790,6 +790,10 @@ ModuleMethodSpec( 快照映射和无返回值能力。Dispatcher 在 provider 返回边界记录 `module.contract.result_mismatch` 与期望形状；该阶段只观测和告警，不拒绝旧插件、不改写返回值， 也不把业务对象类型强行导入动态调度器。未知第三方方法继续完全使用 legacy fallback。 +- 2026-08-24 将已登记的 `first_non_empty` 与 `ordered_list_merge` 接入同步、异步 dispatcher 的统一 + provider 决策函数，避免契约字段只存在于快照而运行时仍执行另一套隐式算法。未登记方法和仍声明 + `legacy` 的宿主/插件能力继续保留原签名接力、列表合并、异常隔离和短路行为；旧 provider 的签名或 + 结果偏差仍只诊断，不拒绝插件加载和执行。 #### ARCH-241：Event Contract Registry diff --git a/tests/test_module_invocation_dispatcher.py b/tests/test_module_invocation_dispatcher.py index 56d448af8..b1aed83a9 100644 --- a/tests/test_module_invocation_dispatcher.py +++ b/tests/test_module_invocation_dispatcher.py @@ -167,6 +167,81 @@ def test_system_signature_relay_passes_previous_result() -> None: assert dispatcher.dispatch("execute") == {"value": 2} +def test_first_non_empty_contract_stops_legacy_signature_relay() -> None: + """显式首个非空契约不得再把结果交给后续宿主 provider 改写。""" + class FirstModule: + """返回首个识别结果的宿主模块。""" + + @staticmethod + def get_name() -> str: + """返回测试模块名。""" + return "第一识别源" + + @staticmethod + def get_priority() -> int: + """返回第一优先级。""" + return 10 + + @staticmethod + def recognize_media() -> str: + """返回首个非空识别结果。""" + return "first" + + class RelayCompatibleModule: + """模拟可接受上一结果的旧式宿主模块。""" + + @staticmethod + def get_name() -> str: + """返回测试模块名。""" + return "旧式接力源" + + @staticmethod + def get_priority() -> int: + """返回第二优先级。""" + return 20 + + @staticmethod + def recognize_media(previous: str) -> str: + """若被调用则改写上一结果。""" + return f"relayed:{previous}" + + dispatcher, _, _, _ = _dispatcher( + modules=[RelayCompatibleModule(), FirstModule()] + ) + + assert dispatcher.dispatch("recognize_media") == "first" + + +def test_ordered_list_contract_bypasses_legacy_signature_relay() -> None: + """显式列表聚合契约应按原参数调用并保留 provider 顺序。""" + class SearchModule: + """区分原参数调用与旧式结果接力的搜索模块。""" + + @staticmethod + def get_name() -> str: + """返回测试模块名。""" + return "系统搜索源" + + @staticmethod + def get_priority() -> int: + """返回稳定优先级。""" + return 10 + + @staticmethod + def search_medias(previous: list | None = None) -> list[str]: + """原参数调用返回系统结果，接力调用返回可检测哨兵。""" + return ["relayed"] if previous is not None else ["system"] + + dispatcher, _, _, _ = _dispatcher( + plugins={ + ("P1", "插件一"): {"search_medias": lambda: ["plugin"]}, + }, + modules=[SearchModule()], + ) + + assert dispatcher.dispatch("search_medias") == ["plugin", "system"] + + def test_module_exception_uses_error_policy_and_continues() -> None: """普通异常应交给错误策略，后续空结果模块仍可继续运行。""" def broken(): @@ -209,6 +284,44 @@ async def test_async_dispatch_awaits_coroutines_and_offloads_sync_functions() -> assert offloaded == [sync_module.execute] +@pytest.mark.asyncio +async def test_async_ordered_list_contract_uses_same_aggregation_policy() -> None: + """异步 dispatcher 应与同步路径共享显式列表聚合语义。""" + class SearchModule: + """提供异步路径下可识别调用方式的同步 provider。""" + + @staticmethod + def get_name() -> str: + """返回测试模块名。""" + return "异步系统搜索源" + + @staticmethod + def get_priority() -> int: + """返回稳定优先级。""" + return 10 + + @staticmethod + def search_medias(previous: list | None = None) -> list[str]: + """原参数调用返回系统结果，接力调用返回可检测哨兵。""" + return ["relayed"] if previous is not None else ["system"] + + async def plugin_search() -> list[str]: + """返回插件搜索结果。""" + return ["plugin"] + + dispatcher, _, _, _ = _dispatcher( + plugins={ + ("P1", "插件一"): {"search_medias": plugin_search}, + }, + modules=[SearchModule()], + ) + + assert await dispatcher.async_dispatch("search_medias") == [ + "plugin", + "system", + ] + + def test_plugin_non_mapping_module_decl_is_reported_and_skipped() -> None: """插件把方法表声明成 list 时走错误策略，且不影响后续健康插件。""" dispatcher, plugin_error, _, _ = _dispatcher(