Merge pull request #6414 from InfinityPacer/codex/replay/g4

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):
     """

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 (
@@ -49,7 +49,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

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 await run_in_threadpool_to_completion(method_func, *args, **kwargs)
-            return method_func(*args, **kwargs)
 
     def get_plugin_ids(self) -> List[str]:
         """

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6451,
+  "edge_count": 6452,
-  "edge_sha256": "9075d8717e384580cd6a41bc36685438770db4a8d8f18c57e6c494f32937113a",
+  "edge_sha256": "fb19022826b8955ed7639c669083f8d1238795bf997e23b28f588d885610aee0",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -5626,6 +5626,7 @@
     "app.runtime.extensions.plugin_manager -> app.foundation.version",
     "app.runtime.extensions.plugin_manager -> app.runtime",
     "app.runtime.extensions.plugin_manager -> app.runtime.events",
+    "app.runtime.extensions.plugin_manager -> app.runtime.execution",
     "app.runtime.extensions.plugin_manager -> app.runtime.extensions",
     "app.runtime.extensions.plugin_manager -> app.runtime.extensions.plugin",
     "app.runtime.extensions.plugin_manager -> app.runtime.extensions.plugin.access",

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

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 == []