From 727e7342cf2003d27ad9fde328018573c1a8b99e Mon Sep 17 00:00:00 2001 From: Todd Dailey <251794714+twidtwid@users.noreply.github.com> Date: Sun, 13 Sep 2026 12:36:03 -0700 Subject: [PATCH] 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 | 8 +-- hermes_cli/lifecycle.py | 9 +++ hermes_cli/plugins.py | 6 ++ hermes_cli/plugins_dispatch.py | 65 ++++++++++++++++++---- tests/gateway/test_bot_loop_guard.py | 12 +++- tests/gateway/test_pre_gateway_dispatch.py | 39 ++++++++++++- tests/hermes_cli/test_plugins.py | 51 +++++++++++++++++ 7 files changed, 169 insertions(+), 21 deletions(-) diff --git a/gateway/run_inbound.py b/gateway/run_inbound.py index 4ca82c962c..72e126098c 100644 --- a/gateway/run_inbound.py +++ b/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 diff --git a/hermes_cli/lifecycle.py b/hermes_cli/lifecycle.py index bdd11a7a97..4ea636a1e5 100644 --- a/hermes_cli/lifecycle.py +++ b/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: diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 529ad7f61f..342fde3c6c 100644 --- a/hermes_cli/plugins.py +++ b/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) diff --git a/hermes_cli/plugins_dispatch.py b/hermes_cli/plugins_dispatch.py index 837dcb3fc5..be1950cfe0 100644 --- a/hermes_cli/plugins_dispatch.py +++ b/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, ())) diff --git a/tests/gateway/test_bot_loop_guard.py b/tests/gateway/test_bot_loop_guard.py index 8b9052bf83..91f4111e66 100644 --- a/tests/gateway/test_bot_loop_guard.py +++ b/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) diff --git a/tests/gateway/test_pre_gateway_dispatch.py b/tests/gateway/test_pre_gateway_dispatch.py index 7c6ccf5e21..3a2243eed9 100644 --- a/tests/gateway/test_pre_gateway_dispatch.py +++ b/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() diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 2711512344..fe5d394919 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/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"}]