fix(gateway): await async pre_gateway_dispatch callbacks on the event loop

`GatewayInboundMixin._hm_pre_gateway_dispatch_hook` was a plain `def` calling
the sync `hermes_cli.lifecycle.invoke_hook` from the async `_hm_admit_event`,
so an `async def pre_gateway_dispatch` callback was resolved through
`resolve_plugin_command_result` on a helper thread with its own loop: the
gateway loop blocked for the callback's whole duration and any loop-bound await
(an `asyncio.Event` set by a loop task, a loop-bound aiohttp session,
`asyncio.to_thread`) could never complete, failing at 30s.

Add `PluginManager.ainvoke_hook` (+ `hermes_cli.plugins.ainvoke_hook` /
`hermes_cli.lifecycle.ainvoke_hook`): same payload narrowing (shared
`_hook_callback_kwargs`), observer + isolation semantics and result contract as
`invoke_hook`, but awaitable results are awaited on the caller's loop.
`pre_gateway_dispatch` stays intentionally unbounded. The inbound hook becomes
`async def` and `_hm_admit_event` awaits it; the sync `invoke_hook` is untouched
for every other caller. Existing tests that stubbed the hook synchronously are
adapted to the async seam.

Fixes #110241
Salvages #110265

(cherry picked from commit 22bb10d305c7992f43bbfdf1481ad0ef0015b2b7)

gateway/run_inbound.py

@@ -63,15 +63,15 @@ def strip_discord_triggering_note(event: Any, message_text: Any) -> Any:
 class GatewayInboundMixin:
     """Inbound message pipeline (_handle_message, text/media preparation, durable-turn markers, plugin injection) for GatewayRunner."""
 
-    def _hm_pre_gateway_dispatch_hook(
+    async def _hm_pre_gateway_dispatch_hook(
         self, event: "MessageEvent", source: SessionSource
     ) -> Optional["MessageEvent"]:
         """Run the ``pre_gateway_dispatch`` plugin hook; None = drop, else the (maybe rewritten) event.
         Results: ``{"action": "skip"}`` → drop; ``{"action": "rewrite", "text"}`` → replace ``event.text``;
         ``allow``/None → normal dispatch. Runs BEFORE auth so plugins can handle unauthorized senders."""
         try:
-            from hermes_cli.lifecycle import invoke_hook as _invoke_hook
-            _hook_results = _invoke_hook(
+            from hermes_cli.lifecycle import ainvoke_hook as _ainvoke_hook
+            _hook_results = await _ainvoke_hook(
                 "pre_gateway_dispatch", event=event, gateway=self,
                 # getattr: bare-runner tests build GatewayRunner via object.__new__ without __init__.
                 session_store=getattr(self, "session_store", None),
@@ -222,7 +222,7 @@ class GatewayInboundMixin:
         # scale-to-zero: only real user-originated inbound stamps the last-inbound clock;
         # counting internal/system events would keep a genuinely idle gateway awake.
         self._scale_to_zero_note_real_inbound()
-        event = self._hm_pre_gateway_dispatch_hook(event, source)
+        event = await self._hm_pre_gateway_dispatch_hook(event, source)
         if event is None:
             return None
         source = event.source

hermes_cli/lifecycle.py

@@ -29,6 +29,15 @@ def invoke_hook(hook_name: str, **kwargs: Any) -> List[Any]:
     return _plugin_hooks(hook_name, **kwargs)
 
 
+async def ainvoke_hook(hook_name: str, **kwargs: Any) -> List[Any]:
+    """:func:`invoke_hook` for callers on an event loop: same observers-then-plugins
+    composition, with ``async def`` plugin callbacks awaited on that loop."""
+    _observe(hook_name, **kwargs)
+    from hermes_cli import plugins
+
+    return await plugins.ainvoke_hook(hook_name, **kwargs)
+
+
 def has_hook(hook_name: str) -> bool:
     """Return whether a first-party observer or plugin consumes a hook."""
     try:

hermes_cli/plugins.py

@@ -1716,6 +1716,12 @@ def invoke_hook(hook_name: str, **kwargs: Any) -> List[Any]:
     return _delivery_manager().invoke_hook(hook_name, **kwargs)
 
 
+async def ainvoke_hook(hook_name: str, **kwargs: Any) -> List[Any]:
+    """:func:`invoke_hook` for callers on an event loop: ``async def`` callbacks are awaited
+    there instead of bridged through a helper thread (see ``PluginManager.ainvoke_hook``)."""
+    return await _delivery_manager().ainvoke_hook(hook_name, **kwargs)
+
+
 def render_system_prompt_sections(session_info: Mapping[str, Any]) -> List[RenderedPluginSystemPromptSection]:
     """Render plugin prompt sections after idempotent plugin discovery."""
     return _ensure_plugins_discovered().render_system_prompt_sections(session_info)

hermes_cli/plugins_dispatch.py

@@ -6,6 +6,7 @@ the origin (tests patch it there) and is looked up lazily.
 
 from __future__ import annotations
 
+import asyncio
 import contextvars
 import copy
 import inspect
@@ -179,7 +180,23 @@ def _hook_uses_callback_timeout(hook_name: str, timeout: float) -> bool:
 
 class PluginDispatchMixin:
     @staticmethod
-    def _invoke_hook_callback(callback: Callable, payload: Dict[str, Any]) -> Any:
+    def _hook_callback_kwargs(callback: Callable, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """The slice of *payload* a callback accepts: everything for ``**kwargs`` (or
+        un-introspectable) callbacks, only declared names for narrow legacy signatures."""
+        try:
+            parameters = inspect.signature(callback).parameters
+        except (TypeError, ValueError):
+            return dict(payload)  # no introspectable signature
+        if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in parameters.values()):
+            return dict(payload)
+        keyword_kinds = {inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY}
+        return {
+            name: value for name, value in payload.items()
+            if name in parameters and parameters[name].kind in keyword_kinds
+        }
+
+    @classmethod
+    def _invoke_hook_callback(cls, callback: Callable, payload: Dict[str, Any]) -> Any:
         """Invoke a hook while withholding additive fields from narrow legacy callbacks.
 
         An ``async def`` callback returns a coroutine; resolve it the way plugin slash commands
@@ -187,17 +204,7 @@ class PluginDispatchMixin:
         plugin's body never runs (#12449).
         """
         from hermes_cli.plugins import resolve_plugin_command_result
-        try:
-            parameters = inspect.signature(callback).parameters
-        except (TypeError, ValueError):
-            return resolve_plugin_command_result(callback(**payload))  # no introspectable signature
-        if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in parameters.values()):
-            return resolve_plugin_command_result(callback(**payload))
-        keyword_kinds = {inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY}
-        return resolve_plugin_command_result(callback(**{
-            name: value for name, value in payload.items()
-            if name in parameters and parameters[name].kind in keyword_kinds
-        }))
+        return resolve_plugin_command_result(callback(**cls._hook_callback_kwargs(callback, payload)))
 
     def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]:
         """Call all callbacks for *hook_name*; return their non-``None`` results.
@@ -473,6 +480,40 @@ class PluginDispatchMixin:
         """Return True when at least one callback is registered for a hook."""
         return bool(self._hooks.get(hook_name))
 
+    async def ainvoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]:
+        """:meth:`invoke_hook` for callers that are already on an event loop.
+
+        Same payload narrowing, per-callback isolation and result contract. The difference is
+        where an ``async def`` callback runs: here it is awaited on the caller's own loop, so a
+        callback that awaits anything scheduled on that loop can make progress. Through the
+        sync path it runs on a helper thread while the caller blocks in ``done.wait()`` — on the
+        gateway that stalls the whole event loop for the callback's duration. Sync callbacks
+        run inline. Bounded hooks keep ``plugins.hook_callback_timeout`` via ``asyncio.wait_for``
+        (the coroutine is cancelled, not abandoned); a timed-out ``pre_tool_call`` fails closed.
+        """
+        from hermes_cli.plugins import _resolve_hook_callback_timeout
+        if hook_name != "gateway_platform_event":
+            kwargs.setdefault("telemetry_schema_version", OBSERVER_SCHEMA_VERSION)
+        results: List[Any] = []
+        timeout = _resolve_hook_callback_timeout()
+        use_timeout = _hook_uses_callback_timeout(hook_name, timeout)
+        fail_closed = hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS
+        for cb in self._hooks.get(hook_name, []):
+            callback_name = getattr(cb, "__name__", repr(cb))
+            try:
+                ret = cb(**self._hook_callback_kwargs(cb, kwargs))
+                if inspect.isawaitable(ret):
+                    ret = await (asyncio.wait_for(ret, timeout) if use_timeout else ret)
+                if ret is not None:
+                    results.append(ret)
+            except asyncio.TimeoutError:
+                logger.warning("Hook '%s' callback %s timed out after %.0fs", hook_name, callback_name, timeout)
+                if fail_closed:  # policy hook: fail closed with a block directive
+                    results.append({"action": "block", "message": _PRE_TOOL_CALL_TIMEOUT_BLOCK_MESSAGE})
+            except Exception as exc:
+                logger.warning("Hook '%s' callback %s raised: %s", hook_name, callback_name, exc)
+        return results
+
     def iter_hook_callbacks(self, hook_name: str) -> tuple[Callable, ...]:
         """Return a stable snapshot of callbacks registered for a hook."""
         return tuple(self._hooks.get(hook_name, ()))

tests/gateway/test_bot_loop_guard.py

@@ -97,7 +97,9 @@ async def test_ingress_gate_counts_an_authorized_bot_once_and_drops_it_when_refu
 
     runner = object.__new__(GatewayRunner)
     runner._scale_to_zero_note_real_inbound = lambda: None
-    runner._hm_pre_gateway_dispatch_hook = lambda event, source: event
+    async def _passthrough_hook(event, source):  # the inbound path awaits the hook
+        return event
+    runner._hm_pre_gateway_dispatch_hook = _passthrough_hook
     runner._is_user_authorized_for_source = lambda source, **kw: True
     admitted = []
     runner._admit_bot_message = lambda source: admitted.append(source.user_id) or source.user_id != BOT_B
@@ -136,7 +138,9 @@ async def test_busy_path_counts_a_bot_message_once_before_steering(monkeypatch,
     assert steer.await_count == 1
 
     runner._scale_to_zero_note_real_inbound = lambda: None
-    runner._hm_pre_gateway_dispatch_hook = lambda event, source: event
+    async def _passthrough_hook(event, source):  # the inbound path awaits the hook
+        return event
+    runner._hm_pre_gateway_dispatch_hook = _passthrough_hook
     runner._is_user_authorized_for_source = lambda source, **kw: True
     runner._admit_bot_message = lambda source: pytest.fail("the busy path already charged this event")
     assert (await runner._hm_admit_event(events[0]))[0] is events[0]
@@ -158,7 +162,9 @@ async def test_routed_bot_traffic_is_metered_by_the_transport_profiles_policy(tm
     runner._principal_authorized = lambda *a, **kw: True
     runner._adapter_profile_for_source = lambda source: "transport"
     runner._scale_to_zero_note_real_inbound = lambda: None
-    runner._hm_pre_gateway_dispatch_hook = lambda event, source: event
+    async def _passthrough_hook(event, source):  # the inbound path awaits the hook
+        return event
+    runner._hm_pre_gateway_dispatch_hook = _passthrough_hook
 
     def _routed_bot(i: int) -> MessageEvent:
         source = _bot(BOT_A)

tests/gateway/test_pre_gateway_dispatch.py

@@ -102,13 +102,14 @@ async def test_hook_fires_without_session_store_attribute(monkeypatch):
 
     seen = {}
 
-    def _fake_hook(name, **kwargs):
+    async def _fake_hook(name, **kwargs):
         if name == "pre_gateway_dispatch":
             seen["session_store"] = kwargs.get("session_store", "MISSING")
             return [{"action": "skip", "reason": "plugin-handled"}]
         return []
 
-    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", _fake_hook)
+    # The inbound path awaits the hook, so the seam is the async entry point.
+    monkeypatch.setattr("hermes_cli.plugins.ainvoke_hook", _fake_hook)
 
     runner, adapter = _make_runner(Platform.WHATSAPP)
     del runner.session_store
@@ -118,3 +119,37 @@ async def test_hook_fires_without_session_store_attribute(monkeypatch):
     # Hook actually fired (skip short-circuited before auth) with a None store.
     assert seen == {"session_store": None}
     adapter.send.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_async_hook_callback_is_awaited_on_the_gateway_loop(monkeypatch):
+    """An ``async def`` pre_gateway_dispatch callback is awaited on the gateway's own loop.
+
+    Regression: the inbound path called the sync ``invoke_hook``, which (since #109196) runs an
+    async callback on a helper thread while the calling loop blocks in ``done.wait()``. A
+    callback that awaits anything scheduled on the gateway loop could never complete, and every
+    message stalled the loop for the callback's whole duration. Here the callback waits for a
+    sibling task on the same loop to release it; that is only possible if the hook is awaited
+    in place.
+    """
+    import asyncio
+
+    _clear_auth_env(monkeypatch)
+    gate = asyncio.Event()
+
+    async def _hook(name, **kwargs):
+        assert name == "pre_gateway_dispatch"
+        await gate.wait()
+        return [{"action": "skip", "reason": "gated"}]
+
+    monkeypatch.setattr("hermes_cli.plugins.ainvoke_hook", _hook)
+
+    async def _release():
+        await asyncio.sleep(0)
+        gate.set()
+
+    runner, adapter = _make_runner(Platform.WHATSAPP)
+    asyncio.create_task(_release())
+    result = await asyncio.wait_for(runner._handle_message(_make_event("hi")), timeout=5)
+    assert result is None
+    adapter.send.assert_not_awaited()

tests/hermes_cli/test_plugins.py

@@ -2876,3 +2876,54 @@ class TestDispatchToolWithoutCliRef:
             assert calls[0][1].get("parent_agent") is None
         finally:
             registry.deregister("_test_dispatch_probe")
+
+
+class TestAsyncHookOnCallerLoop:
+    """``ainvoke_hook`` awaits ``async def`` callbacks on the caller's own event loop.
+
+    #109196 made async callbacks run under ``invoke_hook`` by bridging them through a helper
+    thread; the caller blocks in ``done.wait()`` until the callback finishes. For a hook fired
+    from a coroutine (``pre_gateway_dispatch`` on the gateway loop) that stalls the loop, and a
+    callback that awaits anything scheduled on that loop can never complete. The async twin keeps
+    the callback on the caller's loop.
+    """
+
+    def test_callback_that_needs_the_caller_loop_completes(self):
+        import asyncio
+
+        mgr = PluginManager()
+
+        async def driver():
+            gate = asyncio.Event()
+
+            async def async_hook(**kwargs):
+                await gate.wait()  # only a sibling task on THIS loop can release it
+                return {"action": "allow"}
+
+            async def release():
+                await asyncio.sleep(0)
+                gate.set()
+
+            mgr._hooks.setdefault("pre_gateway_dispatch", []).append(async_hook)
+            asyncio.create_task(release())
+            return await asyncio.wait_for(
+                mgr.ainvoke_hook("pre_gateway_dispatch", event="e", gateway="g"), timeout=5)
+
+        assert asyncio.run(driver()) == [{"action": "allow"}]
+
+    def test_narrow_legacy_signature_still_gets_only_its_fields(self):
+        """Payload narrowing is shared with ``invoke_hook``: a callback declaring only ``event``
+        must not receive the additive ``gateway`` / ``telemetry_schema_version`` fields."""
+        import asyncio
+
+        mgr = PluginManager()
+
+        def narrow(event):
+            return {"seen": event}
+
+        async def narrow_async(event):
+            return {"seen_async": event}
+
+        mgr._hooks.setdefault("pre_gateway_dispatch", []).extend([narrow, narrow_async])
+        results = asyncio.run(mgr.ainvoke_hook("pre_gateway_dispatch", event="e", gateway="g"))
+        assert results == [{"seen": "e"}, {"seen_async": "e"}]