diff --git a/app/runtime/execution.py b/app/runtime/execution.py index 4a0ab2f55..48839734f 100644 --- a/app/runtime/execution.py +++ b/app/runtime/execution.py @@ -21,6 +21,31 @@ async def run_in_threadpool( return await run_sync(context.run, func, *args) +async def run_in_threadpool_to_completion( + func: Callable[..., Any], + *args: Any, + **kwargs: Any, +) -> Any: + """在线程调用取得终态后传播取消，避免提前释放仍在使用的执行容量。""" + worker_task = asyncio.create_task(run_in_threadpool(func, *args, **kwargs)) + cancellation: asyncio.CancelledError | None = None + while not worker_task.done(): + try: + await asyncio.wait({worker_task}) + except asyncio.CancelledError as error: + cancellation = cancellation or error + continue + try: + result = worker_task.result() + except Exception as error: + if cancellation is not None: + raise cancellation from error + raise + if cancellation is not None: + raise cancellation + return result + + def retry(ExceptionToCheck: Any, tries: int = 3, delay: int = 3, backoff: int = 2, logger: Any = None): """ diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py index 0b7ab5fb7..cf5463dcc 100644 --- a/app/runtime/extensions/module/dispatcher.py +++ b/app/runtime/extensions/module/dispatcher.py @@ -7,7 +7,7 @@ from collections.abc import Callable, Mapping from typing import Any, Protocol, cast from app.foundation.reflection import ObjectUtils -from app.runtime.execution import run_in_threadpool +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 ( @@ -50,7 +50,7 @@ class ModuleInvocationDispatcher: plugin_error_handler: ModuleErrorHandler, system_error_handler: ModuleErrorHandler, rate_limit_handler: ModuleErrorHandler, - async_function_runner: AsyncFunctionRunner = run_in_threadpool, + async_function_runner: AsyncFunctionRunner = run_in_threadpool_to_completion, ) -> None: """保存模块目录和策略回调，不主动发现或创建任何运行时资源。""" self._module_catalog = module_catalog diff --git a/app/runtime/extensions/plugin_manager.py b/app/runtime/extensions/plugin_manager.py index 746b3a00d..c505f18bb 100644 --- a/app/runtime/extensions/plugin_manager.py +++ b/app/runtime/extensions/plugin_manager.py @@ -1,4 +1,4 @@ -import asyncio +import inspect import posixpath import threading from contextlib import contextmanager @@ -13,6 +13,7 @@ from app.schemas.plugin import PluginInstance, PluginRuntimeStatus from app.foundation.crypto import RSAUtils from app.foundation.singleton import Singleton from app.foundation.version import compare_version +from app.runtime.execution import run_in_threadpool_to_completion from app.runtime.log import logger from app.runtime.observability import observe_compat_facade from app.runtime.settings import RuntimeSettingsCompat @@ -937,7 +938,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): async def async_run_plugin_method(self, pid: str, method: str, *args, **kwargs) -> Any: """ - 异步运行插件方法 + 异步运行插件方法，同步实现经受控线程入口执行 :param pid: 插件ID :param method: 方法名 :param args: 参数 @@ -949,10 +950,9 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): if not hasattr(plugin, method): return None method_func = getattr(plugin, method) - if asyncio.iscoroutinefunction(method_func): + if inspect.iscoroutinefunction(method_func): return await method_func(*args, **kwargs) - else: - return method_func(*args, **kwargs) + return await run_in_threadpool_to_completion(method_func, *args, **kwargs) def get_plugin_ids(self) -> List[str]: """ diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 766042d30..9ea776895 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6478, - "edge_sha256": "d1c26c72787d07b411f6e047863bfb135005be8d3cf2212f2f03e0d8e0326283", + "edge_count": 6451, + "edge_sha256": "9075d8717e384580cd6a41bc36685438770db4a8d8f18c57e6c494f32937113a", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -1510,8 +1510,6 @@ "app.api.context -> app.application.subscription.delete", "app.api.context -> app.application.subscription.identity", "app.api.context -> app.application.subscription.mutation", - "app.api.context -> app.runtime", - "app.api.context -> app.runtime.tasks", "app.api.context -> app.startup", "app.api.context -> app.startup.context", "app.api.dependencies.agent -> app.api", @@ -1601,7 +1599,6 @@ "app.api.dependencies.subscription -> app.runtime", "app.api.dependencies.subscription -> app.runtime.events", "app.api.dependencies.subscription -> app.runtime.log", - "app.api.dependencies.subscription -> app.runtime.tasks", "app.api.dependencies.subscription -> app.schemas", "app.api.dependencies.subscription -> app.schemas.types", "app.api.dependencies.subscription -> app.startup", @@ -1685,7 +1682,6 @@ "app.api.endpoints.anthropic -> app.agent", "app.api.endpoints.anthropic -> app.agent.runtime_loader", "app.api.endpoints.anthropic -> app.api", - "app.api.endpoints.anthropic -> app.api.context", "app.api.endpoints.anthropic -> app.api.endpoints", "app.api.endpoints.anthropic -> app.api.endpoints.openai", "app.api.endpoints.anthropic -> app.api.openai_utils", @@ -1693,8 +1689,6 @@ "app.api.endpoints.anthropic -> app.api.presentation.sse", "app.api.endpoints.anthropic -> app.application", "app.api.endpoints.anthropic -> app.application.configuration", - "app.api.endpoints.anthropic -> app.runtime", - "app.api.endpoints.anthropic -> app.runtime.tasks", "app.api.endpoints.anthropic -> app.schemas", "app.api.endpoints.anthropic -> app.schemas.openai", "app.api.endpoints.auth -> app.api", @@ -1839,7 +1833,6 @@ "app.api.endpoints.history -> app.runtime.config", "app.api.endpoints.history -> app.runtime.log", "app.api.endpoints.history -> app.runtime.progress", - "app.api.endpoints.history -> app.runtime.tasks", "app.api.endpoints.history -> app.schemas", "app.api.endpoints.history -> app.schemas.common", "app.api.endpoints.history -> app.schemas.history", @@ -1952,7 +1945,6 @@ "app.api.endpoints.message -> app.adapters.web.security", "app.api.endpoints.message -> app.adapters.web.security.access", "app.api.endpoints.message -> app.api", - "app.api.endpoints.message -> app.api.context", "app.api.endpoints.message -> app.api.dependencies", "app.api.endpoints.message -> app.api.dependencies.agent", "app.api.endpoints.message -> app.api.dependencies.auth", @@ -1969,7 +1961,6 @@ "app.api.endpoints.message -> app.runtime.extensions", "app.api.endpoints.message -> app.runtime.extensions.service_config", "app.api.endpoints.message -> app.runtime.log", - "app.api.endpoints.message -> app.runtime.tasks", "app.api.endpoints.message -> app.schemas", "app.api.endpoints.message -> app.schemas.message", "app.api.endpoints.message -> app.schemas.response", @@ -2038,14 +2029,11 @@ "app.api.endpoints.openai -> app.agent.contracts", "app.api.endpoints.openai -> app.agent.runtime_loader", "app.api.endpoints.openai -> app.api", - "app.api.endpoints.openai -> app.api.context", "app.api.endpoints.openai -> app.api.openai_utils", "app.api.endpoints.openai -> app.api.presentation", "app.api.endpoints.openai -> app.api.presentation.sse", "app.api.endpoints.openai -> app.application", "app.api.endpoints.openai -> app.application.configuration", - "app.api.endpoints.openai -> app.runtime", - "app.api.endpoints.openai -> app.runtime.tasks", "app.api.endpoints.openai -> app.schemas", "app.api.endpoints.openai -> app.schemas.openai", "app.api.endpoints.openai -> app.schemas.types", @@ -2060,7 +2048,6 @@ "app.api.endpoints.plugin -> app.adapters.web.security", "app.api.endpoints.plugin -> app.adapters.web.security.access", "app.api.endpoints.plugin -> app.api", - "app.api.endpoints.plugin -> app.api.context", "app.api.endpoints.plugin -> app.api.dependencies", "app.api.endpoints.plugin -> app.api.dependencies.auth", "app.api.endpoints.plugin -> app.api.dependencies.plugin", @@ -2082,7 +2069,6 @@ "app.api.endpoints.plugin -> app.runtime.extensions.plugin", "app.api.endpoints.plugin -> app.runtime.extensions.plugin.contracts", "app.api.endpoints.plugin -> app.runtime.log", - "app.api.endpoints.plugin -> app.runtime.tasks", "app.api.endpoints.plugin -> app.schemas", "app.api.endpoints.plugin -> app.schemas.common", "app.api.endpoints.plugin -> app.schemas.exception", @@ -2136,7 +2122,6 @@ "app.api.endpoints.site -> app.adapters.web.security", "app.api.endpoints.site -> app.adapters.web.security.access", "app.api.endpoints.site -> app.api", - "app.api.endpoints.site -> app.api.context", "app.api.endpoints.site -> app.api.dependencies", "app.api.endpoints.site -> app.api.dependencies.auth", "app.api.endpoints.site -> app.api.dependencies.site", @@ -2160,7 +2145,6 @@ "app.api.endpoints.site -> app.domain.site", "app.api.endpoints.site -> app.runtime", "app.api.endpoints.site -> app.runtime.log", - "app.api.endpoints.site -> app.runtime.tasks", "app.api.endpoints.site -> app.schemas", "app.api.endpoints.site -> app.schemas.common", "app.api.endpoints.site -> app.schemas.response", @@ -2196,7 +2180,6 @@ "app.api.endpoints.subscribe -> app.adapters.web.security", "app.api.endpoints.subscribe -> app.adapters.web.security.access", "app.api.endpoints.subscribe -> app.api", - "app.api.endpoints.subscribe -> app.api.context", "app.api.endpoints.subscribe -> app.api.dependencies", "app.api.endpoints.subscribe -> app.api.dependencies.auth", "app.api.endpoints.subscribe -> app.api.dependencies.subscription", @@ -2218,7 +2201,6 @@ "app.api.endpoints.subscribe -> app.domain.metainfo", "app.api.endpoints.subscribe -> app.runtime", "app.api.endpoints.subscribe -> app.runtime.events", - "app.api.endpoints.subscribe -> app.runtime.tasks", "app.api.endpoints.subscribe -> app.schemas", "app.api.endpoints.subscribe -> app.schemas.common", "app.api.endpoints.subscribe -> app.schemas.event", @@ -2365,12 +2347,9 @@ "app.api.endpoints.webhook -> app.adapters.web.security", "app.api.endpoints.webhook -> app.adapters.web.security.access", "app.api.endpoints.webhook -> app.api", - "app.api.endpoints.webhook -> app.api.context", "app.api.endpoints.webhook -> app.api.response", "app.api.endpoints.webhook -> app.chain", "app.api.endpoints.webhook -> app.chain.webhook", - "app.api.endpoints.webhook -> app.runtime", - "app.api.endpoints.webhook -> app.runtime.tasks", "app.api.endpoints.webhook -> app.schemas", "app.api.endpoints.webhook -> app.schemas.response", "app.api.endpoints.workflow -> app.adapters", @@ -2576,8 +2555,6 @@ "app.application.mediaserver -> app.schemas.mediaserver", "app.application.mediaserver -> app.schemas.system", "app.application.mediaserver -> app.schemas.types", - "app.application.messaging.agent -> app.runtime", - "app.application.messaging.agent -> app.runtime.tasks", "app.application.messaging.agent -> app.schemas", "app.application.messaging.agent -> app.schemas.types", "app.application.messaging.chat -> app.application", @@ -3642,6 +3619,7 @@ "app.db.oper.passkey -> app.db.base", "app.db.oper.passkey -> app.db.models", "app.db.oper.passkey -> app.db.models.passkey", + "app.db.oper.passkey -> app.db.uow", "app.db.oper.plugindata -> app.db", "app.db.oper.plugindata -> app.db.base", "app.db.oper.plugindata -> app.db.models", @@ -4322,7 +4300,6 @@ "app.modules.imdb.api -> app.runtime.cache", "app.modules.imdb.api -> app.runtime.log", "app.modules.imdb.api -> app.runtime.settings", - "app.modules.imdb.api -> app.runtime.tasks", "app.modules.indexer -> app.application", "app.modules.indexer -> app.application.site", "app.modules.indexer -> app.application.site.health", @@ -6017,8 +5994,6 @@ "app.startup.context -> app.application.subscription.identity", "app.startup.context -> app.application.subscription.mutation", "app.startup.context -> app.application.workflow", - "app.startup.context -> app.runtime", - "app.startup.context -> app.runtime.tasks", "app.startup.database -> app.adapters", "app.startup.database -> app.adapters.system", "app.startup.database -> app.adapters.system.backup", @@ -6082,7 +6057,6 @@ "app.startup.lifecycle -> app.runtime.log", "app.startup.lifecycle -> app.runtime.settings", "app.startup.lifecycle -> app.runtime.state", - "app.startup.lifecycle -> app.runtime.tasks", "app.startup.lifecycle -> app.runtime.topology", "app.startup.lifecycle -> app.startup", "app.startup.lifecycle -> app.startup.cache_initializer", @@ -6197,7 +6171,6 @@ "app.startup.modules_initializer -> app.runtime.observability", "app.startup.modules_initializer -> app.runtime.settings", "app.startup.modules_initializer -> app.runtime.state", - "app.startup.modules_initializer -> app.runtime.tasks", "app.startup.modules_initializer -> app.runtime.thread", "app.startup.modules_initializer -> app.scheduler", "app.startup.modules_initializer -> app.schemas", @@ -6495,7 +6468,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 800, + "module_count": 799, "modules": [ "app", "app.adapters", @@ -7188,7 +7161,6 @@ "app.runtime.scheduling", "app.runtime.settings", "app.runtime.state", - "app.runtime.tasks", "app.runtime.thread", "app.runtime.topology", "app.scheduler", diff --git a/tests/test_plugin_manager_async_methods.py b/tests/test_plugin_manager_async_methods.py new file mode 100644 index 000000000..dcd8965f1 --- /dev/null +++ b/tests/test_plugin_manager_async_methods.py @@ -0,0 +1,71 @@ +"""插件管理器异步方法的执行边界回归。""" + +from types import SimpleNamespace +from typing import Iterator +from unittest.mock import AsyncMock + +import pytest + +from app.foundation.singleton import Singleton +from app.sdk.plugins import PluginManager + + +@pytest.fixture +def plugin_manager() -> Iterator[PluginManager]: + """构造隔离的插件管理器实例。""" + Singleton._instances.pop((PluginManager, (), frozenset()), None) + manager = PluginManager() + yield manager + Singleton._instances.pop((PluginManager, (), frozenset()), None) + + +@pytest.mark.asyncio +async def test_async_run_plugin_method_offloads_sync_plugin_method( + plugin_manager: PluginManager, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """异步入口不得在事件循环内直接执行同步插件方法。""" + calls: list[tuple[str, int]] = [] + + def sync_method(value: int) -> int: + calls.append(("sync", value)) + return value + 1 + + plugin_manager.running_plugins["DemoPlugin"] = SimpleNamespace( + sync_method=sync_method, + ) + worker = AsyncMock(return_value=2) + monkeypatch.setattr( + "app.runtime.extensions.plugin_manager.run_in_threadpool_to_completion", + worker, + ) + + assert await plugin_manager.async_run_plugin_method( + "DemoPlugin", "sync_method", 1 + ) == 2 + worker.assert_awaited_once_with(sync_method, 1) + assert calls == [] + + +@pytest.mark.asyncio +async def test_async_run_plugin_method_keeps_async_plugin_method_on_loop( + plugin_manager: PluginManager, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """原生协程插件方法继续直接等待，不额外占用同步 worker。""" + async def async_method(value: int) -> int: + return value + 1 + + plugin_manager.running_plugins["DemoPlugin"] = SimpleNamespace( + async_method=async_method, + ) + worker = AsyncMock() + monkeypatch.setattr( + "app.runtime.extensions.plugin_manager.run_in_threadpool_to_completion", + worker, + ) + + assert await plugin_manager.async_run_plugin_method( + "DemoPlugin", "async_method", 1 + ) == 2 + worker.assert_not_awaited() diff --git a/tests/test_runtime_execution.py b/tests/test_runtime_execution.py new file mode 100644 index 000000000..532ec5c7a --- /dev/null +++ b/tests/test_runtime_execution.py @@ -0,0 +1,90 @@ +"""运行时同步 worker 的取消与容量合同回归。""" + +import asyncio +import threading + +import pytest +from anyio.to_thread import current_default_thread_limiter + +from app.runtime.execution import run_in_threadpool_to_completion + + +@pytest.mark.asyncio +async def test_threadpool_capacity_is_held_until_cancelled_call_finishes() -> None: + """调用方取消后，执行令牌必须由真实同步调用持有到终态。""" + limiter = current_default_thread_limiter() + original_capacity = limiter.total_tokens + release = threading.Event() + first_started = threading.Event() + second_started = threading.Event() + + def blocking_call(started: threading.Event) -> None: + started.set() + release.wait() + + limiter.total_tokens = 1 + first = asyncio.create_task( + run_in_threadpool_to_completion(blocking_call, first_started) + ) + second = None + try: + while not first_started.is_set(): + await asyncio.sleep(0) + + first.cancel() + await asyncio.sleep(0) + first.cancel() + await asyncio.sleep(0) + + assert first.done() is False + assert limiter.borrowed_tokens == 1 + + second = asyncio.create_task( + run_in_threadpool_to_completion(blocking_call, second_started) + ) + await asyncio.sleep(0.01) + assert second_started.is_set() is False + + release.set() + with pytest.raises(asyncio.CancelledError): + await first + await second + finally: + release.set() + if not first.done(): + await asyncio.gather(first, return_exceptions=True) + if second is not None and not second.done(): + await asyncio.gather(second, return_exceptions=True) + limiter.total_tokens = original_capacity + + +@pytest.mark.asyncio +async def test_cancelled_threadpool_call_preserves_worker_failure_as_cause() -> None: + """调用方取消优先返回，线程终态异常仍保留为诊断原因。""" + loop = asyncio.get_running_loop() + previous_handler = loop.get_exception_handler() + loop_errors: list[dict] = [] + release = threading.Event() + started = threading.Event() + + def failing_call() -> None: + started.set() + release.wait() + raise ValueError("worker failed") + + task = asyncio.create_task(run_in_threadpool_to_completion(failing_call)) + while not started.is_set(): + await asyncio.sleep(0) + + task.cancel() + release.set() + + loop.set_exception_handler(lambda _loop, context: loop_errors.append(context)) + try: + with pytest.raises(asyncio.CancelledError) as error_info: + await task + await asyncio.sleep(0) + finally: + loop.set_exception_handler(previous_handler) + assert isinstance(error_info.value.__cause__, ValueError) + assert loop_errors == []