From 5a99664b36a7f11d8f4f534aa8fe34904399e097 Mon Sep 17 00:00:00 2001
From: MervinPraison <454862+MervinPraison@users.noreply.github.com>
Date: Fri, 23 Oct 2026 10:02:58 +0000
Subject: [PATCH 1/6] fix: await async def plugin hooks instead of dropping the
 coroutine (fixes #5964)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../praisonaiagents/plugins/manager.py        | 76 +++++++++++----
 .../praisonaiagents/plugins/plugin.py         |  5 +-
 .../unit/plugins/test_async_plugin_hooks.py   | 92 +++++++++++++++++++
 3 files changed, 151 insertions(+), 22 deletions(-)
 create mode 100644 src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py

diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py
index e95e66b6b5..1288c8c3be 100644
--- a/src/praisonai-agents/praisonaiagents/plugins/manager.py
+++ b/src/praisonai-agents/praisonaiagents/plugins/manager.py
@@ -542,7 +542,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 {})
@@ -1013,6 +1014,38 @@ def _emit_plugin_lifecycle(event: "HookEvent", plugin: Plugin) -> None:
         logger.debug("Plugin lifecycle hook failed", exc_info=True)
 
 
+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. 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=None)
+    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 +1082,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 +1122,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 +1137,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 +1146,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 +1162,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 +1171,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 +1187,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 +1202,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 +1234,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 +1248,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 +1265,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 +1281,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 +1299,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 +1309,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 +1319,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 +1330,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 +1344,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 +1402,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..a8224122b6
--- /dev/null
+++ b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
@@ -0,0 +1,92 @@
+"""``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}

From 13b33e55860d475419d658e17cf26fdb55236fd1 Mon Sep 17 00:00:00 2001
From: MervinPraison <454862+MervinPraison@users.noreply.github.com>
Date: Fri, 23 Oct 2026 10:19:16 +0000
Subject: [PATCH 2/6] fix: bound async plugin hooks with a deadline and await
 async on_init/on_config/on_shutdown

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../praisonaiagents/plugins/manager.py        | 21 ++++---
 .../unit/plugins/test_async_plugin_hooks.py   | 57 +++++++++++++++++++
 2 files changed, 70 insertions(+), 8 deletions(-)

diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py
index 1288c8c3be..ef2217c3c9 100644
--- a/src/praisonai-agents/praisonaiagents/plugins/manager.py
+++ b/src/praisonai-agents/praisonaiagents/plugins/manager.py
@@ -262,11 +262,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}")
@@ -298,7 +297,7 @@ def register(self, plugin: Plugin) -> bool:
                 self._enabled[info.name] = self._default_enabled_for_registration(info.name)
                 
                 # Initialize plugin
-                plugin.on_init({})
+                _ResolvingPlugin(plugin).on_init({})
 
                 # Make the initialization observable to the hook system
                 # too. ``plugin.on_init`` is the plugin's *own* callback;
@@ -331,7 +330,7 @@ def unregister(self, name: str) -> bool:
             self._enabled.pop(name, None)
 
         try:
-            plugin.on_shutdown()
+            _ResolvingPlugin(plugin).on_shutdown()
         except Exception as e:
             logger.warning(f"Error during plugin shutdown: {e}")
 
@@ -1014,17 +1013,23 @@ 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. Sync hook
-    return values pass through unchanged.
+    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=None)
+        return run_coroutine_from_any_context(value, timeout=_ASYNC_HOOK_TIMEOUT)
     return value
 
 
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
index a8224122b6..26d7caf9f7 100644
--- a/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
+++ b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
@@ -90,3 +90,60 @@ def test_execute_hook_awaits_async_hook_sync_and_async_paths():
     }
     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]

From b85cb817ae35e02d5ef14e9e558071af15b5a7cf Mon Sep 17 00:00:00 2001
From: MervinPraison <454862+MervinPraison@users.noreply.github.com>
Date: Fri, 23 Oct 2026 11:03:50 +0000
Subject: [PATCH 3/6] fix: run plugin on_init outside the manager lock so async
 on_init cannot deadlock

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../praisonaiagents/plugins/manager.py        | 60 ++++++++++---------
 .../unit/plugins/test_async_plugin_hooks.py   | 33 ++++++++++
 2 files changed, 65 insertions(+), 28 deletions(-)

diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py
index ef2217c3c9..2d6bce63be 100644
--- a/src/praisonai-agents/praisonaiagents/plugins/manager.py
+++ b/src/praisonai-agents/praisonaiagents/plugins/manager.py
@@ -295,22 +295,27 @@ def register(self, plugin: Plugin) -> bool:
 
                 self._plugins[info.name] = plugin
                 self._enabled[info.name] = self._default_enabled_for_registration(info.name)
-                
-                # Initialize plugin
-                _ResolvingPlugin(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
             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({})
+
+            # 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}")
+            return False
     
     def unregister(self, name: str) -> bool:
         """
@@ -844,21 +849,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
 
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
index 26d7caf9f7..1891c3cd4a 100644
--- a/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
+++ b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
@@ -147,3 +147,36 @@ async def hang():
         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

From 4b956e72fc273b4c28462b3eda25d19323d525ea Mon Sep 17 00:00:00 2001
From: MervinPraison <454862+MervinPraison@users.noreply.github.com>
Date: Fri, 23 Oct 2026 11:06:58 +0000
Subject: [PATCH 4/6] fix: roll back plugin registration when on_init fails

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../praisonaiagents/plugins/manager.py        |  7 +++++-
 .../unit/plugins/test_async_plugin_hooks.py   | 23 +++++++++++++++++++
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py
index 2d6bce63be..68d1d622ed 100644
--- a/src/praisonai-agents/praisonaiagents/plugins/manager.py
+++ b/src/praisonai-agents/praisonaiagents/plugins/manager.py
@@ -315,8 +315,13 @@ def register(self, plugin: Plugin) -> bool:
             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)
             return False
-    
+
     def unregister(self, name: str) -> bool:
         """
         Unregister a plugin.
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
index 1891c3cd4a..ec4bd0d0ac 100644
--- a/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
+++ b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
@@ -180,3 +180,26 @@ async def main():
     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

From b3c55dba4a1a2299f59d82baf567b3b64bd1e63d Mon Sep 17 00:00:00 2001
From: MervinPraison <454862+MervinPraison@users.noreply.github.com>
Date: Sat, 24 Oct 2026 00:58:40 +0000
Subject: [PATCH 5/6] fix: keep plugins out of dispatch until on_init succeeds

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../praisonaiagents/plugins/manager.py        | 11 ++++-
 .../unit/plugins/test_async_plugin_hooks.py   | 47 +++++++++++++++++++
 2 files changed, 57 insertions(+), 1 deletion(-)

diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py
index 68d1d622ed..7755a0066e 100644
--- a/src/praisonai-agents/praisonaiagents/plugins/manager.py
+++ b/src/praisonai-agents/praisonaiagents/plugins/manager.py
@@ -293,8 +293,10 @@ 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)
+                self._enabled[info.name] = False
+                default_enabled = self._default_enabled_for_registration(info.name)
             except Exception as e:
                 logger.error(f"Failed to register plugin: {e}")
                 return False
@@ -304,6 +306,13 @@ def register(self, plugin: Plugin) -> bool:
         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() made during on_init is kept.
+                self._enabled[info.name] = default_enabled or self._enabled.get(info.name, False)
+
             # 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
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
index ec4bd0d0ac..42a33614f2 100644
--- a/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
+++ b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
@@ -203,3 +203,50 @@ async def on_init(self, context):
 
     # 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

From 676abc8cb08894940e02f602c995dd1185cb7c77 Mon Sep 17 00:00:00 2001
From: MervinPraison <454862+MervinPraison@users.noreply.github.com>
Date: Sat, 24 Oct 2026 01:01:46 +0000
Subject: [PATCH 6/6] fix: keep an explicit disable() made during plugin
 on_init

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../praisonaiagents/plugins/manager.py           | 14 ++++++++++++--
 .../unit/plugins/test_async_plugin_hooks.py      | 16 ++++++++++++++++
 2 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py
index 7755a0066e..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}.
@@ -296,6 +299,7 @@ def register(self, plugin: Plugin) -> bool:
                 # Reserve the name but keep it out of dispatch until on_init succeeds.
                 self._plugins[info.name] = plugin
                 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}")
@@ -310,8 +314,10 @@ def register(self, plugin: Plugin) -> bool:
                 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() made during on_init is kept.
-                self._enabled[info.name] = default_enabled or self._enabled.get(info.name, 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;
@@ -329,6 +335,7 @@ def register(self, plugin: Plugin) -> bool:
                 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:
@@ -347,6 +354,7 @@ def unregister(self, name: str) -> bool:
                 return False
             del self._plugins[name]
             self._enabled.pop(name, None)
+            self._initializing.discard(name)
 
         try:
             _ResolvingPlugin(plugin).on_shutdown()
@@ -365,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
     
@@ -379,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:
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
index 42a33614f2..2311806567 100644
--- a/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
+++ b/src/praisonai-agents/tests/unit/plugins/test_async_plugin_hooks.py
@@ -250,3 +250,19 @@ def before_tool(self, tool_name, args):
     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