diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py index e95e66b6b5..d1717c6e4f 100644 --- a/src/praisonai-agents/praisonaiagents/plugins/manager.py +++ b/src/praisonai-agents/praisonaiagents/plugins/manager.py @@ -143,6 +143,9 @@ class PluginManager: def __init__(self, disabled: Optional[bool] = None): self._plugins: Dict[str, Plugin] = {} self._enabled: Dict[str, bool] = {} + # Names whose on_init is running and that have not been explicitly + # enabled/disabled yet; register() applies the default only to these. + self._initializing: set = set() self._single_file_plugins: Dict[str, Dict[str, Any]] = {} # WordPress-style plugins # Per-plugin option maps sourced from the unified project config, # delivered to each plugin's on_config hook. {plugin_name: options}. @@ -262,11 +265,10 @@ def apply_plugin_options(self) -> int: if self._enabled.get(name, False) and name in self._plugin_options ] for name, plugin, options in targets: - on_config = getattr(plugin, "on_config", None) - if not callable(on_config): + if not callable(getattr(plugin, "on_config", None)): continue try: - on_config(options) + _ResolvingPlugin(plugin).on_config(options) delivered += 1 except Exception as e: logger.warning(f"Plugin '{name}' on_config failed: {e}") @@ -294,25 +296,48 @@ def register(self, plugin: Plugin) -> bool: logger.warning(f"Plugin '{info.name}' disabled: {reason}") return False + # Reserve the name but keep it out of dispatch until on_init succeeds. self._plugins[info.name] = plugin - self._enabled[info.name] = self._default_enabled_for_registration(info.name) - - # Initialize plugin - plugin.on_init({}) - - # Make the initialization observable to the hook system - # too. ``plugin.on_init`` is the plugin's *own* callback; - # ON_INIT is the runtime event other listeners - # (metrics/audit plugins, gateways) subscribe to. Without - # this emit the enum member was declared but never fired. - _emit_plugin_init(plugin) - - logger.info(f"Registered plugin: {info.name} v{info.version}") - return True + self._enabled[info.name] = False + self._initializing.add(info.name) + default_enabled = self._default_enabled_for_registration(info.name) except Exception as e: logger.error(f"Failed to register plugin: {e}") return False - + + # Initialize outside the lock: an ``async def on_init`` may run on + # another thread and call back into this manager. + try: + _ResolvingPlugin(plugin).on_init({}) + + with self._lock: + if self._plugins.get(info.name) is not plugin: + logger.warning(f"Plugin '{info.name}' was unregistered during on_init") + return False + # Publish; an explicit enable()/disable() made during on_init is kept. + if info.name in self._initializing: + self._initializing.discard(info.name) + self._enabled[info.name] = default_enabled + + # Make the initialization observable to the hook system + # too. ``plugin.on_init`` is the plugin's *own* callback; + # ON_INIT is the runtime event other listeners + # (metrics/audit plugins, gateways) subscribe to. Without + # this emit the enum member was declared but never fired. + _emit_plugin_init(plugin) + + logger.info(f"Registered plugin: {info.name} v{info.version}") + return True + except Exception as e: + logger.error(f"Failed to register plugin: {e}") + # Roll back so a failed plugin is not left half-registered. + with self._lock: + if self._plugins.get(info.name) is plugin: + del self._plugins[info.name] + self._enabled.pop(info.name, None) + self._initializing.discard(info.name) + return False + def unregister(self, name: str) -> bool: """ Unregister a plugin. @@ -329,9 +354,10 @@ def unregister(self, name: str) -> bool: return False del self._plugins[name] self._enabled.pop(name, None) + self._initializing.discard(name) try: - plugin.on_shutdown() + _ResolvingPlugin(plugin).on_shutdown() except Exception as e: logger.warning(f"Error during plugin shutdown: {e}") @@ -347,6 +373,7 @@ def enable(self, name: str) -> bool: with self._lock: if name in self._enabled: self._enabled[name] = True + self._initializing.discard(name) return True return False @@ -361,6 +388,7 @@ def disable(self, name: str) -> bool: if name not in self._enabled: return False self._enabled[name] = False + self._initializing.discard(name) try: self.unwire_from_hook_registry(name) except Exception as e: @@ -542,7 +570,8 @@ def execute_hook( continue try: - method = getattr(plugin, hook.value, None) + # Resolving view: an ``async def`` hook is awaited, not dropped. + method = getattr(_ResolvingPlugin(plugin), hook.value, None) if method: if hook == PluginHook.BEFORE_TOOL: result = method(args[0], args[1] if len(args) > 1 else {}) @@ -844,21 +873,20 @@ def discover_entry_points(self) -> int: return 0 loaded = 0 - # Use lock to ensure thread-safe discovery of multiple entry points. - # register() also acquires the lock but that's fine (RLock allows re-entry). - with self._lock: - for ep in entry_points: - try: - plugin_cls = ep.load() - if callable(plugin_cls): - plugin = plugin_cls() - if self.register(plugin): - loaded += 1 - logger.info(f"Loaded entry_point plugin: {ep.name}") - else: - logger.warning(f"Entry point {ep.name} is not callable") - except Exception as e: - logger.error(f"Failed to load plugin from entry point {ep.name}: {e}") + # register() is thread-safe on its own; holding the lock here would + # keep it held while a plugin's (possibly async) on_init runs. + for ep in entry_points: + try: + plugin_cls = ep.load() + if callable(plugin_cls): + plugin = plugin_cls() + if self.register(plugin): + loaded += 1 + logger.info(f"Loaded entry_point plugin: {ep.name}") + else: + logger.warning(f"Entry point {ep.name} is not callable") + except Exception as e: + logger.error(f"Failed to load plugin from entry point {ep.name}: {e}") return loaded @@ -1013,6 +1041,44 @@ def _emit_plugin_lifecycle(event: "HookEvent", plugin: Plugin) -> None: logger.debug("Plugin lifecycle hook failed", exc_info=True) +# Deadline for an ``async def`` plugin hook; matches HookRunner's default +# per-hook timeout so a hung hook is cancelled and frees its worker thread. +_ASYNC_HOOK_TIMEOUT = 60.0 + + +def _resolve_awaitable(value: Any) -> Any: + """Drive a coroutine returned by an ``async def`` plugin hook to completion. + + Hook adapters run on the HookRunner's worker thread (no running loop), so + the coroutine runs on a private loop there; from a thread with a running + loop the shared bridge offloads it instead of deadlocking. The coroutine is + bounded by ``_ASYNC_HOOK_TIMEOUT``. Sync hook return values pass through + unchanged. + """ + if asyncio.iscoroutine(value): + from ..utils.async_bridge import run_coroutine_from_any_context + return run_coroutine_from_any_context(value, timeout=_ASYNC_HOOK_TIMEOUT) + return value + + +class _ResolvingPlugin: + """View of a plugin whose method calls resolve ``async def`` hooks.""" + + __slots__ = ("_plugin",) + + def __init__(self, plugin: Plugin): + self._plugin = plugin + + def __getattr__(self, name: str) -> Any: + attr = getattr(self._plugin, name) + if not callable(attr): + return attr + + def call(*args, **kwargs): + return _resolve_awaitable(attr(*args, **kwargs)) + return call + + def _write_back(data, attr: str, new_value: dict) -> None: """Write a plugin's returned dict back onto the hook payload. @@ -1049,6 +1115,9 @@ def _adapt_plugin_hooks(plugin: Plugin) -> Iterator[Tuple["HookEvent", Callable] from .plugin import GuardrailBlocked, PluginDecision base = Plugin + # Adapters call the plugin through this view so ``async def`` hooks are + # awaited instead of returning an unawaited coroutine. + resolving_plugin = _ResolvingPlugin(plugin) def _as_decision(result: Any) -> Optional[HookResult]: """Coerce a plugin's return value into a denying ``HookResult``. @@ -1086,7 +1155,7 @@ def _permission_result(value: Optional[bool], reason: str) -> HookResult: return HookResult.allow() if _overrides("before_agent"): - def before_agent_hook(data, _p=plugin): + def before_agent_hook(data, _p=resolving_plugin): try: new_prompt = _p.before_agent(getattr(data, "prompt", ""), {"agent_name": getattr(data, "agent_name", None)}) @@ -1101,7 +1170,7 @@ def before_agent_hook(data, _p=plugin): yield HookEvent.BEFORE_AGENT, before_agent_hook if _overrides("after_agent"): - def after_agent_hook(data, _p=plugin): + def after_agent_hook(data, _p=resolving_plugin): new_resp = _p.after_agent(getattr(data, "response", ""), {"agent_name": getattr(data, "agent_name", None)}) if isinstance(new_resp, str) and hasattr(data, "response"): @@ -1110,7 +1179,7 @@ def after_agent_hook(data, _p=plugin): yield HookEvent.AFTER_AGENT, after_agent_hook if _overrides("before_llm"): - def before_llm_hook(data, _p=plugin): + def before_llm_hook(data, _p=resolving_plugin): messages = getattr(data, "messages", []) try: result = _p.before_llm(messages, {"model": getattr(data, "model", "")}) @@ -1126,7 +1195,7 @@ def before_llm_hook(data, _p=plugin): yield HookEvent.BEFORE_LLM, before_llm_hook if _overrides("after_llm"): - def after_llm_hook(data, _p=plugin): + def after_llm_hook(data, _p=resolving_plugin): new_resp = _p.after_llm(getattr(data, "response", ""), {"tokens_used": getattr(data, "tokens_used", 0)}) if isinstance(new_resp, str) and hasattr(data, "response"): @@ -1135,7 +1204,7 @@ def after_llm_hook(data, _p=plugin): yield HookEvent.AFTER_LLM, after_llm_hook if _overrides("before_tool"): - def before_tool_hook(data, _p=plugin): + def before_tool_hook(data, _p=resolving_plugin): args = getattr(data, "tool_input", {}) or {} try: new_args = _p.before_tool(getattr(data, "tool_name", ""), args) @@ -1151,7 +1220,7 @@ def before_tool_hook(data, _p=plugin): yield HookEvent.BEFORE_TOOL, before_tool_hook if _overrides("after_tool"): - def after_tool_hook(data, _p=plugin): + def after_tool_hook(data, _p=resolving_plugin): try: new_result = _p.after_tool(getattr(data, "tool_name", ""), getattr(data, "tool_output", None)) @@ -1166,7 +1235,7 @@ def after_tool_hook(data, _p=plugin): yield HookEvent.AFTER_TOOL, after_tool_hook if _overrides("before_tool_definitions"): - def before_tool_definitions_hook(data, _p=plugin): + def before_tool_definitions_hook(data, _p=resolving_plugin): defs = getattr(data, "tool_definitions", []) or [] try: new_defs = _p.before_tool_definitions(defs) @@ -1198,7 +1267,7 @@ def _message_payload(data): return payload if _overrides("before_message"): - def before_message_hook(data, _p=plugin): + def before_message_hook(data, _p=resolving_plugin): try: new = _p.before_message(_message_payload(data)) except GuardrailBlocked as e: @@ -1212,7 +1281,7 @@ def before_message_hook(data, _p=plugin): yield HookEvent.MESSAGE_RECEIVED, before_message_hook if _overrides("after_message"): - def after_message_hook(data, _p=plugin): + def after_message_hook(data, _p=resolving_plugin): try: new = _p.after_message(_message_payload(data)) except GuardrailBlocked as e: @@ -1229,14 +1298,14 @@ def after_message_hook(data, _p=plugin): yield HookEvent.MESSAGE_SENDING, after_message_hook if _overrides("message_sent"): - def message_sent_hook(data, _p=plugin): + def message_sent_hook(data, _p=resolving_plugin): payload = _message_payload(data) _p.message_sent(payload) return HookResult.allow() yield HookEvent.MESSAGE_SENT, message_sent_hook if _overrides("message_undelivered"): - def message_undelivered_hook(data, _p=plugin): + def message_undelivered_hook(data, _p=resolving_plugin): payload = _message_payload(data) payload["error"] = getattr(data, "error", "") payload["notice_delivered"] = getattr(data, "notice_delivered", False) @@ -1245,14 +1314,14 @@ def message_undelivered_hook(data, _p=plugin): yield HookEvent.MESSAGE_UNDELIVERED, message_undelivered_hook if _overrides("on_permission_ask"): - def on_permission_ask_hook(data, _p=plugin): + def on_permission_ask_hook(data, _p=resolving_plugin): target = getattr(data, "tool_name", "") or getattr(data, "target", "") reason = getattr(data, "reason", "") or "Permission requested" return _permission_result(_p.on_permission_ask(target, reason), reason) yield HookEvent.ON_PERMISSION_ASK, on_permission_ask_hook if _overrides("on_config"): - def on_config_hook(data, _p=plugin): + def on_config_hook(data, _p=resolving_plugin): config = getattr(data, "config", None) if not isinstance(config, dict): config = getattr(data, "extra", {}) or {} @@ -1263,7 +1332,7 @@ def on_config_hook(data, _p=plugin): yield HookEvent.ON_CONFIG, on_config_hook if _overrides("on_auth"): - def on_auth_hook(data, _p=plugin): + def on_auth_hook(data, _p=resolving_plugin): auth_type = getattr(data, "auth_type", "") or "" credentials = getattr(data, "credentials", None) or {} new_creds = _p.on_auth(auth_type, credentials) @@ -1273,7 +1342,7 @@ def on_auth_hook(data, _p=plugin): yield HookEvent.ON_AUTH, on_auth_hook if _overrides("session_start"): - def session_start_hook(data, _p=plugin): + def session_start_hook(data, _p=resolving_plugin): _p.session_start({ "source": getattr(data, "source", None), "session_name": getattr(data, "session_name", None), @@ -1283,7 +1352,7 @@ def session_start_hook(data, _p=plugin): yield HookEvent.SESSION_START, session_start_hook if _overrides("session_end"): - def session_end_hook(data, _p=plugin): + def session_end_hook(data, _p=resolving_plugin): _p.session_end({ "reason": getattr(data, "reason", None), "total_turns": getattr(data, "total_turns", None), @@ -1294,7 +1363,7 @@ def session_end_hook(data, _p=plugin): yield HookEvent.SESSION_END, session_end_hook if _overrides("on_error"): - def on_error_hook(data, _p=plugin): + def on_error_hook(data, _p=resolving_plugin): _p.on_error( getattr(data, "error_type", "") or "", getattr(data, "error_message", "") or "", @@ -1308,7 +1377,7 @@ def on_error_hook(data, _p=plugin): yield HookEvent.ON_ERROR, on_error_hook if _overrides("cli_backend_execute"): - def cli_backend_execute_hook(data, _p=plugin): + def cli_backend_execute_hook(data, _p=resolving_plugin): from ..cli_backend.debug import redact_command _p.cli_backend_execute({ "agent_name": getattr(data, "agent_name", None), @@ -1366,7 +1435,7 @@ def observe_hook(data, _m=method): # ``kanban_task_created``) are treated as genuine overrides. if parent is not None and own is parent: continue - yield event, _make_observe_hook(method) + yield event, _make_observe_hook(getattr(resolving_plugin, event.value)) # Global plugin manager instance diff --git a/src/praisonai-agents/praisonaiagents/plugins/plugin.py b/src/praisonai-agents/praisonaiagents/plugins/plugin.py index 7c12db0149..adc93cb7cd 100644 --- a/src/praisonai-agents/praisonaiagents/plugins/plugin.py +++ b/src/praisonai-agents/praisonaiagents/plugins/plugin.py @@ -177,8 +177,9 @@ class Plugin(ABC): Plugins can extend agent functionality by implementing hook methods that are called at various points in the - agent lifecycle. - + agent lifecycle. Hook methods may be ``def`` or ``async def``; + coroutines returned by async hooks are awaited by the dispatcher. + Example: class MyPlugin(Plugin): @property diff --git a/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py new file mode 100644 index 0000000000..2311806567 --- /dev/null +++ b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py @@ -0,0 +1,268 @@ +"""``async def`` plugin hooks are awaited, not silently dropped (#5964).""" + +import asyncio +import os +import time +import warnings + +from praisonaiagents.plugins.plugin import ( + GuardrailBlocked, + Plugin, + PluginHook, + PluginInfo, +) +from praisonaiagents.plugins.manager import PluginManager +from praisonaiagents.hooks.registry import HookRegistry +from praisonaiagents.hooks.runner import HookRunner +from praisonaiagents.hooks.types import HookEvent +from praisonaiagents.hooks.events import BeforeToolInput, SessionStartInput + + +class AsyncToolPlugin(Plugin): + def __init__(self): + self.sessions = [] + + @property + def info(self): + return PluginInfo( + name="async_tool", + hooks=[PluginHook.BEFORE_TOOL, PluginHook.SESSION_START], + ) + + async def before_tool(self, tool_name, args): + await asyncio.sleep(0) + if tool_name == "rm": + raise GuardrailBlocked("rm is not allowed") + return {**args, "injected": True} + + async def session_start(self, context): + await asyncio.sleep(0) + self.sessions.append(context["source"]) + + +def _tool_input(tool_name="bash"): + return BeforeToolInput( + session_id="s", cwd=os.getcwd(), event_name=HookEvent.BEFORE_TOOL, + timestamp=str(time.time()), agent_name="a", + tool_name=tool_name, tool_input={"cmd": "ls"}, + ) + + +def _runner(plugin): + reg = HookRegistry() + mgr = PluginManager() + mgr.register(plugin) + mgr.wire_into_hook_registry(reg) + return HookRunner(registry=reg, cwd=os.getcwd()) + + +def test_async_transforming_hook_is_awaited_via_runner(): + runner = _runner(AsyncToolPlugin()) + data = _tool_input() + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + runner.execute_sync(HookEvent.BEFORE_TOOL, data) + assert data.tool_input == {"cmd": "ls", "injected": True} + + +def test_async_hook_guardrail_block_propagates(): + runner = _runner(AsyncToolPlugin()) + results = runner.execute_sync(HookEvent.BEFORE_TOOL, _tool_input("rm")) + assert any(r.output and r.output.is_denied() for r in results) + + +def test_async_observer_hook_runs_inside_running_loop(): + plugin = AsyncToolPlugin() + runner = _runner(plugin) + data = SessionStartInput( + session_id="s", cwd=os.getcwd(), event_name=HookEvent.SESSION_START, + timestamp=str(time.time()), source="startup", + ) + asyncio.run(runner.execute(HookEvent.SESSION_START, data)) + assert plugin.sessions == ["startup"] + + +def test_execute_hook_awaits_async_hook_sync_and_async_paths(): + mgr = PluginManager() + mgr.register(AsyncToolPlugin()) + assert mgr.execute_hook(PluginHook.BEFORE_TOOL, "bash", {"a": 1}) == { + "a": 1, "injected": True, + } + out = asyncio.run(mgr.async_execute_hook(PluginHook.BEFORE_TOOL, "bash", {"a": 1})) + assert out == {"a": 1, "injected": True} + + +class AsyncLifecyclePlugin(Plugin): + def __init__(self): + self.calls = [] + + @property + def info(self): + return PluginInfo(name="async_lifecycle") + + async def on_init(self, context): + await asyncio.sleep(0) + self.calls.append("init") + + async def on_config(self, config): + await asyncio.sleep(0) + self.calls.append(("config", config)) + return config + + async def on_shutdown(self): + await asyncio.sleep(0) + self.calls.append("shutdown") + + +def test_async_lifecycle_callbacks_are_awaited(): + plugin = AsyncLifecyclePlugin() + mgr = PluginManager() + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + assert mgr.register(plugin) + mgr.enable("async_lifecycle") + mgr.set_plugin_options({"async_lifecycle": {"k": 1}}) + assert mgr.apply_plugin_options() == 1 + assert mgr.unregister("async_lifecycle") + assert plugin.calls == ["init", ("config", {"k": 1}), "shutdown"] + + +def test_hung_async_hook_is_cancelled_at_deadline(monkeypatch): + from praisonaiagents.plugins import manager as manager_mod + + monkeypatch.setattr(manager_mod, "_ASYNC_HOOK_TIMEOUT", 0.05) + cancelled = [] + + async def hang(): + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + cancelled.append(True) + raise + + start = time.time() + try: + manager_mod._resolve_awaitable(hang()) + except (TimeoutError, asyncio.TimeoutError): + pass + assert time.time() - start < 2 + assert cancelled == [True] + + +def test_async_on_init_can_call_back_into_manager_from_running_loop(monkeypatch): + """register() must not hold the manager lock while async on_init runs.""" + from praisonaiagents.plugins import manager as manager_mod + + # A deadlock would otherwise only surface after the 60s deadline. + monkeypatch.setattr(manager_mod, "_ASYNC_HOOK_TIMEOUT", 2.0) + mgr = PluginManager() + + class ReentrantInitPlugin(Plugin): + seen = None + + @property + def info(self): + return PluginInfo(name="reentrant_init") + + async def on_init(self, context): + await asyncio.sleep(0) + # Bridged onto another thread when called from a running loop. + # enable() takes the manager lock. + self.seen = mgr.enable("reentrant_init") + + plugin = ReentrantInitPlugin() + + async def main(): + start = time.monotonic() + assert mgr.register(plugin) is True + return time.monotonic() - start + + elapsed = asyncio.run(main()) + assert plugin.seen is True + assert elapsed < 1.5 + + +def test_failed_async_on_init_rolls_back_registration(): + """A plugin whose on_init fails is not left half-registered.""" + mgr = PluginManager() + + class FailingInitPlugin(Plugin): + @property + def info(self): + return PluginInfo(name="failing_init") + + async def on_init(self, context): + raise RuntimeError("boom") + + assert mgr.register(FailingInitPlugin()) is False + assert mgr.get_plugin("failing_init") is None + + class FixedInitPlugin(FailingInitPlugin): + async def on_init(self, context): + return None + + # A retry is not rejected as "already registered". + assert mgr.register(FixedInitPlugin()) is True + + +def test_plugin_not_dispatched_until_async_on_init_finishes(): + """Concurrent dispatch during a slow on_init must not reach the plugin.""" + import threading + + mgr = PluginManager() + init_started = threading.Event() + release_init = threading.Event() + + class SlowInitPlugin(Plugin): + def __init__(self): + self.ready = False + self.calls_before_ready = 0 + + @property + def info(self): + return PluginInfo(name="slow_init", hooks=[PluginHook.BEFORE_TOOL]) + + async def on_init(self, context): + init_started.set() + await asyncio.get_running_loop().run_in_executor(None, release_init.wait, 5) + self.ready = True + + def before_tool(self, tool_name, args): + if not self.ready: + self.calls_before_ready += 1 + return {**args, "seen": True} + + plugin = SlowInitPlugin() + results = {} + t = threading.Thread(target=lambda: results.setdefault("ok", mgr.register(plugin))) + t.start() + assert init_started.wait(5) + + # During on_init: neither direct dispatch nor registry wiring sees it + # (with no enabled plugins execute_hook passes its first arg through). + assert mgr.execute_hook(PluginHook.BEFORE_TOOL, "bash", {"a": 1}) == "bash" + assert mgr.wire_into_hook_registry(HookRegistry()) == 0 + assert mgr.is_enabled("slow_init") is False + + release_init.set() + t.join(5) + assert results["ok"] is True + assert mgr.is_enabled("slow_init") is True + assert mgr.execute_hook(PluginHook.BEFORE_TOOL, "bash", {"a": 1}) == {"a": 1, "seen": True} + assert plugin.calls_before_ready == 0 + + +def test_disable_during_on_init_is_kept(): + """A plugin that disables itself in on_init stays disabled after register.""" + mgr = PluginManager() + + class SelfDisablingPlugin(Plugin): + @property + def info(self): + return PluginInfo(name="self_disabling", hooks=[PluginHook.BEFORE_TOOL]) + + async def on_init(self, context): + mgr.disable("self_disabling") + + assert mgr.register(SelfDisablingPlugin()) is True + assert mgr.is_enabled("self_disabling") is False