refactor: enforce module aggregation contracts

2026-08-29 12:06:51 +08:00 · 2026-08-24 00:57:19 +08:00
parent 2e5dd2cf0a
commit bec64887d6
3 changed files with 233 additions and 43 deletions
@@ -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:
        """仅把真实超时归入低基数模块超时指标。"""
@@ -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

@@ -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(