diff --git a/agent/shell_hooks.py b/agent/shell_hooks.py
index 424d2d9e6df5e..ca7ac6d6b039f 100644
--- a/agent/shell_hooks.py
+++ b/agent/shell_hooks.py
@@ -445,6 +445,15 @@ def _parse_pre_tool_call(data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     for verb, _, _, payload in _PRE_TOOL_DIALECTS:
         if data.get(verb) == "modify" and isinstance(data.get(payload), dict):
             return {"action": "modify", "args": data[payload]}
+    # Hermes-only escalation to the human-approval gate (#92553). Claude-Code's ``decision:
+    # approve`` means auto-ALLOW, so it is deliberately not mapped onto this.
+    if data.get("action") == "approve":
+        directive: Dict[str, Any] = {"action": "approve"}
+        for key in ("message", "rule_key"):
+            value = data.get(key)
+            if isinstance(value, str) and value.strip():
+                directive[key] = value.strip()
+        return directive
     return None
 
 
diff --git a/gateway/run_inbound.py b/gateway/run_inbound.py
index 4ca82c962c534..72e126098c804 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 @@ async def _hm_admit_event(
         # 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 bdd11a7a9756d..4ea636a1e53fe 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/middleware.py b/hermes_cli/middleware.py
index e5d42c3a2c4f1..1c6ed2ae249a3 100644
--- a/hermes_cli/middleware.py
+++ b/hermes_cli/middleware.py
@@ -159,10 +159,10 @@ def __init__(self, original: BaseException) -> None:
 
 
 def _run_execution_chain(kind: str, terminal_call: Callable[[Any], Any], **kwargs: Any) -> Any:
-    from hermes_cli.plugins import get_plugin_manager
+    from hermes_cli.plugins import _delivery_manager
 
     payload_key = "request" if "request" in kwargs else "args"
-    manager = get_plugin_manager()
+    manager = _delivery_manager()
     callbacks = list(manager._middleware.get(kind, []))
     if not callbacks:
         return terminal_call(kwargs[payload_key])
diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py
index 23dff081afdf3..342fde3c6c86e 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)
@@ -1810,8 +1816,10 @@ def _get_pre_tool_call_directive_details(
 ) -> _PreToolCallDirective:
     """Check ``pre_tool_call`` hooks for ``{"action": "block", "message"}`` (veto; message becomes
     the tool result) or ``{"action": "approve", "message", "rule_key"?}`` (escalate ANY tool to the
-    human-approval gate; ``rule_key`` picks the ``[a]lways`` allowlist grain). First valid directive
-    wins; irrelevant returns are ignored."""
+    human-approval gate; ``rule_key`` picks the ``[a]lways`` allowlist grain). Precedence is
+    ``block`` > ``approve`` > none, not registration order: any plugin's valid veto wins over an
+    earlier plugin's request for human confirmation (#87420); among approves the first valid one
+    wins. Irrelevant returns are ignored."""
     allowed = getattr(_thread_tool_whitelist, "allowed", None)
     if allowed is not None and tool_name not in allowed:
         fmt = getattr(_thread_tool_whitelist, "fmt", "Tool '{tool_name}' denied")
@@ -1823,6 +1831,7 @@ def _get_pre_tool_call_directive_details(
         api_request_id=api_request_id, middleware_trace=list(middleware_trace or []),
     )
     modified_args: Optional[Dict[str, Any]] = None
+    first_approve: Optional[Tuple[Optional[str], Optional[str]]] = None  # (message, rule_key)
     for result in hook_results:
         if not isinstance(result, dict):
             continue
@@ -1843,9 +1852,15 @@ def _get_pre_tool_call_directive_details(
         # A block directive requires a message (it becomes the tool result); approve's is optional.
         if action == "block" and not message:
             continue
-        rule_key = result.get("rule_key") if action == "approve" else None
-        rule_key = (rule_key.strip() or None) if isinstance(rule_key, str) else None
-        return _PreToolCallDirective(action=action, message=message, rule_key=rule_key, modified_args=modified_args)
+        if action == "block":
+            return _PreToolCallDirective(action="block", message=message, modified_args=modified_args)
+        # approve is held back until the whole list has been scanned for a veto.
+        if first_approve is None:
+            rule_key = result.get("rule_key")
+            first_approve = (message, (rule_key.strip() or None) if isinstance(rule_key, str) else None)
+    if first_approve is not None:
+        return _PreToolCallDirective(action="approve", message=first_approve[0], rule_key=first_approve[1],
+                                     modified_args=modified_args)
     return _PreToolCallDirective(modified_args=modified_args)
 
 
diff --git a/hermes_cli/plugins_dispatch.py b/hermes_cli/plugins_dispatch.py
index 55b469fdb5023..d1f1df19edffa 100644
--- a/hermes_cli/plugins_dispatch.py
+++ b/hermes_cli/plugins_dispatch.py
@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import contextvars
 import copy
 import inspect
@@ -50,8 +51,19 @@
 _HOOK_CALLER_THREAD_HOOKS: Set[str] = {"subagent_stop"}
 # After a timeout, suppress the same callback this long so a hung hook cannot pile up threads.
 _HOOK_TIMEOUT_SUPPRESSION_SECONDS = 60.0
+# Live workers a hung callback may accumulate before it is skipped outright (#105223 / #98382).
+_HOOK_MAX_ABANDONED_WORKERS = 3
 _PRE_TOOL_CALL_TIMEOUT_BLOCK_MESSAGE = "pre_tool_call plugin callback timed out or is still running"
 
+
+def _policy_error_block_directive(hook_name: str, cb: Callable, exc: BaseException) -> Dict[str, str]:
+    """Block directive for a fail-closed hook whose callback raised: names the callback and the
+    error (truncated — a hook that embeds tool args in its exception must not grow the tool
+    result) so the operator can tell a crashing guard from a slow one."""
+    callback_name = getattr(cb, "__name__", repr(cb))
+    return {"action": "block",
+            "message": f"{hook_name} plugin callback {callback_name} raised {type(exc).__name__}: {str(exc)[:200]}"}
+
 # System-prompt sections are tightly bounded: they become high-trust prompt bytes charged every turn.
 SYSTEM_PROMPT_SECTION_POSITIONS = frozenset({"after_memory"})
 DEFAULT_SYSTEM_PROMPT_SECTION_MAX_CHARS = 4_000
@@ -168,25 +180,31 @@ 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:
-        """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
-        are (loop-safe), otherwise the bare coroutine object is appended to the results and the
-        plugin's body never runs (#12449).
-        """
-        from hermes_cli.plugins import resolve_plugin_command_result
+    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 resolve_plugin_command_result(callback(**payload))  # no introspectable signature
+            return dict(payload)  # no introspectable signature
         if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in parameters.values()):
-            return resolve_plugin_command_result(callback(**payload))
+            return dict(payload)
         keyword_kinds = {inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY}
-        return resolve_plugin_command_result(callback(**{
+        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
+        are (loop-safe), otherwise the bare coroutine object is appended to the results and the
+        plugin's body never runs (#12449).
+        """
+        from hermes_cli.plugins import resolve_plugin_command_result
+        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.
@@ -220,6 +238,8 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]:
                     results.append(ret)
             except (Exception, SystemExit) as exc:
                 self._report_hook_failure(hook_name, cb, kwargs, exc)
+                if fail_closed:  # a guard that raised made no decision: same veto as a timeout
+                    results.append(_policy_error_block_directive(hook_name, cb, exc))
         return results
 
     def _report_hook_failure(
@@ -250,8 +270,9 @@ def _run_hook_callback_bounded(
         self, hook_name: str, cb: Callable, kwargs: Dict[str, Any], timeout: float
     ) -> Any:
         """Run one callback on a daemon worker with a wall-clock cap; ``_HOOK_SKIPPED`` when
-        suppressed, still running, timed out (worker abandoned, never joined), or the worker
-        could not be started. Exceptions propagate."""
+        suppressed, still running for this call id, over the abandoned-worker cap, timed out
+        (worker abandoned, never joined), or the worker could not be started. Exceptions
+        propagate."""
         callback_name = getattr(cb, "__name__", repr(cb))
         # Suppression is a fact about the CALLBACK — a hung one must keep its back-off —
         # so that key stays coarse. The gate must instead tell CONCURRENT CALLS apart.
@@ -260,15 +281,25 @@ def _run_hook_callback_bounded(
         token = object()
         with self._hook_timeout_lock:
             suppressed_until = self._hook_timeout_suppressed_until.get(suppression_key)
-            # A worker abandoned on timeout still holds a thread; a fresh call id must not
-            # start a second one for the same callback, or a hung plugin leaks a thread per call.
-            running = (gate_key in self._hook_running_callbacks
-                       or bool(self._hook_abandoned.get(suppression_key)))
-            if (suppressed_until is not None and suppressed_until > time.monotonic()) or running:
+            if (gate_key in self._hook_running_callbacks
+                    or (suppressed_until is not None and suppressed_until > time.monotonic())):
                 logger.warning(
                     "Hook '%s' callback %s skipped after previous "
                     "timeout or while still running", hook_name, callback_name)
                 return _HOOK_SKIPPED
+            # Workers abandoned on timeout still hold threads. Once the suppression window has
+            # passed, a fresh call id may start a new worker (a hung guard must not fail every
+            # later tool call closed until restart, #105223), but only up to a small cap per
+            # callback — expiring the bookkeeping while the hung worker lives must not leak a
+            # thread per call (#98382). At the cap the callback keeps being skipped (fail-closed
+            # for pre_tool_call) until one of its workers finishes and releases its slot.
+            abandoned = self._hook_abandoned.get(suppression_key)
+            if abandoned and len(abandoned) >= _HOOK_MAX_ABANDONED_WORKERS:
+                logger.warning(
+                    "Hook '%s' callback %s (%s) skipped: %d abandoned worker(s) still running — "
+                    "the plugin is hung; fix or disable it (retried when a worker finishes)",
+                    hook_name, callback_name, getattr(cb, "__module__", "unknown plugin"), len(abandoned))
+                return _HOOK_SKIPPED
             if suppressed_until is not None:
                 self._hook_timeout_suppressed_until.pop(suppression_key, None)
             self._hook_running_callbacks[gate_key] = token
@@ -449,6 +480,44 @@ def has_hook(self, hook_name: str) -> bool:
         """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, SystemExit) as exc:
+                # Same isolation + failure contract as the sync path (#111922 warn-once, #109624
+                # a raising policy guard fails closed).
+                self._report_hook_failure(hook_name, cb, kwargs, exc)
+                if fail_closed:
+                    results.append(_policy_error_block_directive(hook_name, cb, 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/agent/test_shell_hooks.py b/tests/agent/test_shell_hooks.py
index ab376c5b8a54f..e2238b448a303 100644
--- a/tests/agent/test_shell_hooks.py
+++ b/tests/agent/test_shell_hooks.py
@@ -49,7 +49,18 @@ def test_block_claude_code_style(self):
         )
         assert r == {"action": "block", "message": "nope"}
 
-
+    @pytest.mark.parametrize("stdout, expected", [
+        ('{"action": "approve", "message": "  needs a human ", "rule_key": " terminal:rm "}',
+         {"action": "approve", "message": "needs a human", "rule_key": "terminal:rm"}),
+        ('{"action": "approve", "message": "", "rule_key": 7}', {"action": "approve"}),
+        # Claude-Code's ``decision: approve`` means auto-ALLOW, not "ask a human": never mapped.
+        ('{"decision": "approve", "reason": "ok"}', None),
+        ('{"action": "approve", "decision": "block", "reason": "no"}', {"action": "block", "message": "no"}),
+    ])
+    def test_approve_is_parsed_like_the_plugin_directive(self, stdout, expected):
+        """The documented ``approve`` action used to parse to None, so the tool ran with no
+        approval prompt (#92553). It now yields the same shape Python plugins return."""
+        assert shell_hooks._parse_response("pre_tool_call", stdout) == expected
 
     def test_empty_stdout_returns_none(self):
         assert shell_hooks._parse_response("pre_tool_call", "") is None
@@ -201,6 +212,32 @@ def test_block_aggregation_through_plugin_manager(self, tmp_path, monkeypatch):
         )
         assert msg == "blocked-by-shell"
 
+    def test_approve_reaches_the_human_gate_through_plugin_manager(self, tmp_path, monkeypatch):
+        """End to end: a shell hook's approve directive escalates to request_tool_approval with its
+        message and rule_key, and the gate's denial blocks the tool (#92553)."""
+        from hermes_cli import plugins
+
+        script = _write_script(
+            tmp_path, "approve.sh",
+            "#!/usr/bin/env bash\n"
+            'printf \'{"action": "approve", "message": "risky", "rule_key": "terminal:rm"}\\n\'\n',
+        )
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
+        monkeypatch.setenv("HERMES_ACCEPT_HOOKS", "1")
+        plugins._plugin_manager = plugins.PluginManager()
+        cfg = {"hooks": {"pre_tool_call": [{"matcher": "terminal", "command": str(script)}]}}
+        assert len(shell_hooks.register_from_config(cfg, accept_hooks=True)) == 1
+
+        seen = []
+
+        def _gate(tool_name, reason, **kwargs):
+            seen.append((tool_name, reason, kwargs.get("rule_key")))
+            return {"approved": False, "message": "denied by human"}
+
+        monkeypatch.setattr("tools.approval.request_tool_approval", _gate)
+        assert plugins.resolve_pre_tool_block("terminal", {"command": "rm"}) == "denied by human"
+        assert seen == [("terminal", "risky", "terminal:rm")]
+
     def test_matcher_regex_filters_callback(self, tmp_path, monkeypatch):
         """A matcher set to 'terminal' must not fire for 'web_search'."""
         calls = tmp_path / "calls.log"
diff --git a/tests/gateway/test_bot_loop_guard.py b/tests/gateway/test_bot_loop_guard.py
index 8b9052bf83502..91f4111e66fb7 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 7c6ccf5e219cc..3a2243eed9281 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 @@ def _fake_hook(name, **kwargs):
     # 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 c28a26abcc280..2e9c5bc146da3 100644
--- a/tests/hermes_cli/test_plugins.py
+++ b/tests/hermes_cli/test_plugins.py
@@ -31,6 +31,7 @@
     VALID_MIDDLEWARE,
     apply_llm_request_middleware,
     apply_tool_request_middleware,
+    run_llm_execution_middleware,
     run_tool_execution_middleware,
 )
 
@@ -1018,6 +1019,48 @@ def test_invoke_hook_tolerates_mock_managers(self, monkeypatch):
 
         assert plugins_mod.invoke_hook("anything") == ["stubbed"]
 
+    def test_execution_chain_lazily_discovers(self, monkeypatch):
+        """Execution middleware must fire on cold surfaces too (#105827).
+
+        ``run_tool_execution_middleware`` / ``run_llm_execution_middleware`` deliver via
+        ``_run_execution_chain``, which used to read ``get_plugin_manager()._middleware``
+        directly — no lazy discovery — so a registered fail-closed policy gate was silently
+        skipped (fail-open) on surfaces that never ran discovery at startup (query mode
+        ``chat -q``, cron delivery, dashboard, TUI slash workers). The chain must route
+        through ``_delivery_manager()`` like every other delivery entry point (#64178).
+        """
+        fired = []
+        terminal_calls = []
+
+        def _tool_gate(**kw):
+            fired.append(kw.get("tool_name"))
+            return {"denied": True}  # fail-closed: returns without calling next_call
+
+        def _llm_gate(**kw):
+            fired.append("llm")
+            return {"denied": True}
+
+        def _register(m):
+            m._middleware.setdefault("tool_execution", []).append(_tool_gate)
+            m._middleware.setdefault("llm_execution", []).append(_llm_gate)
+
+        mgr = self._fresh_manager(monkeypatch, _register)
+
+        def _terminal(payload):
+            terminal_calls.append(payload)
+            return "terminal-ran"
+
+        tool_result = run_tool_execution_middleware("terminal", {"path": "x"}, _terminal)
+        assert mgr._discovered is True, "execution chain must lazily discover on cold surfaces"
+        assert fired == ["terminal"]
+        assert tool_result == {"denied": True}
+        assert terminal_calls == [], "a fail-closed gate must not be bypassed by terminal execution"
+
+        llm_result = run_llm_execution_middleware({"messages": []}, _terminal)
+        assert fired == ["terminal", "llm"]
+        assert llm_result == {"denied": True}
+        assert terminal_calls == []
+
 
 class TestAsyncHookCallbacks:
     """``async def`` hook callbacks run and their values land in the results (#12449)."""
@@ -1225,6 +1268,25 @@ def exits(**_kwargs):
             assert mgr.invoke_hook("post_tool_call") == ["survived"]
         assert "bounded plugin requested process exit" in caplog.text
 
+    @pytest.mark.parametrize("timeout", [0.0, 1.0], ids=["caller-thread", "bounded-worker"])
+    def test_pre_tool_call_callback_exception_fails_closed(self, monkeypatch, timeout):
+        """A policy callback that raises made no decision: it must block like a timeout does
+        (#109624), on both the caller-thread and the bounded-worker path, and the block message
+        names the callback and the error so a crashing guard is distinguishable from a slow one."""
+        monkeypatch.setattr(
+            "hermes_cli.plugins._resolve_hook_callback_timeout", lambda: timeout
+        )
+
+        def boom(**_kwargs):
+            raise RuntimeError("policy plugin blew up")
+
+        mgr = PluginManager()
+        mgr._hooks["pre_tool_call"] = [boom, lambda **_kw: {"action": "approve"}]
+
+        results = mgr.invoke_hook("pre_tool_call", tool_name="terminal", args={})
+        assert [r.get("action") for r in results] == ["block", "approve"]
+        assert "boom" in results[0]["message"] and "RuntimeError: policy plugin blew up" in results[0]["message"]
+
     def test_hook_callback_timeout_reads_config(self, tmp_path, monkeypatch):
         hermes_home = tmp_path / "hermes_test"
         hermes_home.mkdir(parents=True, exist_ok=True)
@@ -1407,9 +1469,14 @@ def fire():
         hold.set()
         first.join(5.0)
 
-    def test_hung_worker_blocks_new_call_identity_after_suppression(self, monkeypatch):
-        """A worker abandoned on timeout still occupies its callback: a later call with a
-        fresh id must be skipped, not given a second thread (one leak, not one per call)."""
+    def test_hung_worker_caps_new_call_identities_after_suppression(self, monkeypatch, caplog):
+        """Workers abandoned on timeout still occupy their callback: once the suppression window
+        has passed, later calls with fresh ids may start a replacement, but only up to
+        ``_HOOK_MAX_ABANDONED_WORKERS`` live ones — a hung plugin leaks a bounded few threads,
+        never one per call (#98382), and past the cap it is skipped with a warning that names
+        the callback (#105223)."""
+        import hermes_cli.plugins_dispatch as dispatch
+
         monkeypatch.setattr(
             "hermes_cli.plugins._resolve_hook_callback_timeout", lambda: 0.1
         )
@@ -1426,10 +1493,44 @@ def blocker(**_kwargs):
         mgr._hook_timeout_suppression_seconds = 0.0  # isolate the gate from suppression
         mgr._hooks["post_tool_call"] = [blocker]
 
-        assert mgr.invoke_hook("post_tool_call", tool_name="read_file", tool_call_id="call-a") == []
-        assert mgr.invoke_hook("post_tool_call", tool_name="read_file", tool_call_id="call-b") == []
+        with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"):
+            for i in range(dispatch._HOOK_MAX_ABANDONED_WORKERS + 3):
+                assert mgr.invoke_hook("post_tool_call", tool_name="read_file", tool_call_id=f"call-{i}") == []
 
-        assert len(starts) == 1
+        assert len(starts) == dispatch._HOOK_MAX_ABANDONED_WORKERS
+        assert "blocker" in caplog.text and "abandoned worker(s) still running" in caplog.text
+        hold.set()
+
+    def test_hung_worker_does_not_fail_closed_forever(self, monkeypatch):
+        """One never-returning pre_tool_call guard must not block every later tool call until
+        restart: after the suppression window a fresh call id runs a new worker, so a callback
+        that has recovered decides again (#105223)."""
+        import time
+
+        from hermes_cli.plugins import _PRE_TOOL_CALL_TIMEOUT_BLOCK_MESSAGE
+
+        monkeypatch.setattr(
+            "hermes_cli.plugins._resolve_hook_callback_timeout", lambda: 0.1
+        )
+        hold = threading.Event()
+        starts = []
+
+        def guard(**_kwargs):
+            starts.append(1)
+            if len(starts) == 1:
+                hold.wait(timeout=10.0)  # the first fire hangs for good
+            return None  # later fires decide: allow
+
+        mgr = PluginManager()
+        mgr._hook_timeout_suppression_seconds = 0.2
+        mgr._hooks["pre_tool_call"] = [guard]
+
+        blocked = [{"action": "block", "message": _PRE_TOOL_CALL_TIMEOUT_BLOCK_MESSAGE}]
+        assert mgr.invoke_hook("pre_tool_call", tool_name="read_file", tool_call_id="call-a") == blocked
+        assert mgr.invoke_hook("pre_tool_call", tool_name="read_file", tool_call_id="call-b") == blocked  # in window
+        time.sleep(0.3)  # suppression window passes; the first worker is still hung
+        assert mgr.invoke_hook("pre_tool_call", tool_name="read_file", tool_call_id="call-c") == []
+        assert len(starts) == 2
         hold.set()
 
     def test_worker_finishing_at_timeout_does_not_leave_phantom_abandoned_entry(self, monkeypatch):
@@ -1707,6 +1808,41 @@ def test_approve_without_message_is_valid(self, monkeypatch):
         )
         assert get_pre_tool_call_directive("write_file", {}) == ("approve", None)
 
+    def test_later_block_outranks_earlier_approve(self, monkeypatch):
+        """Precedence is block > approve, not registration order: a security plugin's veto must
+        not be shadowed by an earlier plugin's approve (#87420). Under approvals.mode off an
+        approve means no prompt at all, so the veto would otherwise be dropped silently."""
+        from hermes_cli.plugins import _get_pre_tool_call_directive_details
+        monkeypatch.setattr(
+            "hermes_cli.plugins.invoke_hook",
+            lambda hook_name, **kwargs: [
+                {"action": "modify", "args": {"path": "/safe"}},
+                {"action": "approve", "message": "earlier plugin approves", "rule_key": "k"},
+                {"action": "block", "message": "later security plugin blocks"},
+            ],
+        )
+        details = _get_pre_tool_call_directive_details("write_file", {"path": "/unsafe"})
+        assert (details.action, details.message, details.rule_key) == (
+            "block", "later security plugin blocks", None)
+        assert details.modified_args == {"path": "/safe"}  # modify before the veto stays visible
+
+    def test_first_approve_wins_among_approves_and_keeps_later_modify(self, monkeypatch):
+        """Holding approve back for a veto scan must not change which approve wins (first valid,
+        incl. its rule_key) and must keep accumulating modify directives that follow it."""
+        from hermes_cli.plugins import _get_pre_tool_call_directive_details
+        monkeypatch.setattr(
+            "hermes_cli.plugins.invoke_hook",
+            lambda hook_name, **kwargs: [
+                {"action": "block"},  # message-less block is invalid and ignored
+                {"action": "approve", "message": "first", "rule_key": " write_file:ssh "},
+                {"action": "modify", "args": {"content": "fixed"}},
+                {"action": "approve", "message": "second", "rule_key": "write_file:other"},
+            ],
+        )
+        details = _get_pre_tool_call_directive_details("write_file", {"path": "/p"})
+        assert (details.action, details.message, details.rule_key) == ("approve", "first", "write_file:ssh")
+        assert details.modified_args == {"path": "/p", "content": "fixed"}
+
 
 class TestResolvePreToolBlock:
     """Tests for the single dispatch-site chokepoint that resolves a
@@ -2740,3 +2876,38 @@ def test_dispatch_tool_invokes_handler_without_cli_ref(self):
             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_narrow_legacy_signature_still_gets_only_its_fields(self, caplog):
+        """Payload narrowing and failure isolation are shared with ``invoke_hook``: a callback
+        declaring only ``event`` must not receive the additive ``gateway`` /
+        ``telemetry_schema_version`` fields, and a raising callback is reported once and skipped
+        without losing its siblings' results."""
+        import asyncio
+
+        mgr = PluginManager()
+
+        def narrow(event):
+            return {"seen": event}
+
+        async def boom(**_kw):
+            raise RuntimeError("async plugin blew up")
+
+        async def narrow_async(event):
+            return {"seen_async": event}
+
+        mgr._hooks.setdefault("pre_gateway_dispatch", []).extend([narrow, boom, narrow_async])
+        with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"):
+            results = asyncio.run(mgr.ainvoke_hook("pre_gateway_dispatch", event="e", gateway="g"))
+        assert results == [{"seen": "e"}, {"seen_async": "e"}]
+        assert "async plugin blew up" in caplog.text
diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md
index a3d235f332179..f05a308db3c2d 100644
--- a/website/docs/user-guide/features/hooks.md
+++ b/website/docs/user-guide/features/hooks.md
@@ -438,7 +438,7 @@ Payload fields below are the exact event-specific fields supplied by each call s
 
 | Hook | Category | Exact timing and return behavior | Explicit payload fields | Privacy / sensitivity |
 |---|---|---|---|---|
-| [`pre_tool_call`](#pre_tool_call) | Directive/control | Once before execution; first valid `block` or `approve` directive wins, and `modify` returns are shallow-merged into the tool arguments. | `tool_name`, `args`, `task_id`, `session_id`, `tool_call_id`, `turn_id`, `api_request_id`, `middleware_trace` | Raw arguments may contain user content, paths, commands, or secrets. |
+| [`pre_tool_call`](#pre_tool_call) | Directive/control | Once before execution; any valid `block` wins over any `approve` (then the first valid `approve`), and `modify` returns are shallow-merged into the tool arguments. | `tool_name`, `args`, `task_id`, `session_id`, `tool_call_id`, `turn_id`, `api_request_id`, `middleware_trace` | Raw arguments may contain user content, paths, commands, or secrets. |
 | `post_tool_call` | Observer | After blocked, error, or successful result; return ignored. | `tool_name`, `args`, `result`, `task_id`, `session_id`, `tool_call_id`, `turn_id`, `api_request_id`, `duration_ms`, `status`, `error_type`, `error_message`, `middleware_trace` | Result/error text may contain arbitrary tool or user content and secrets. |
 | `transform_tool_result` | Transform | After `post_tool_call`, before conversation append; first string replaces the result. | `tool_name`, `args`, `result`, `task_id`, `session_id`, `tool_call_id`, `turn_id`, `api_request_id`, `duration_ms`, `status`, `error_type`, `error_message` | Exposes the full model-bound result and arguments. |
 | `transform_terminal_output` | Transform | After bounded foreground process capture, before final output limiting; first string replaces output. | `command`, `output`, `returncode`, `task_id`, `env_type` | Command/output may contain credentials. |
@@ -554,7 +554,7 @@ return {"action": "block", "message": "Reason the tool call was blocked"}
 return {"action": "approve", "message": "Why approval is required", "rule_key": "optional:scope"}
 ```
 
-The first valid directive wins (Python plugins registered first, then shell hooks). `block` requires a non-empty `message` and short-circuits the tool with that text as the error returned to the model. `approve` escalates the call to the existing human-approval gate; `message` and `rule_key` are optional, and denial, timeout, or gate error fails closed. Other return values are ignored, so existing observer-only callbacks keep working unchanged.
+Precedence is `block` > `approve` > no directive, regardless of registration order: any plugin's valid `block` wins over an earlier plugin's `approve`, and among `approve` directives the first valid one wins (Python plugins are registered first, then shell hooks). `block` requires a non-empty `message` and short-circuits the tool with that text as the error returned to the model. `approve` escalates the call to the existing human-approval gate; `message` and `rule_key` are optional, and denial, timeout, or gate error fails closed. Other return values are ignored, so existing observer-only callbacks keep working unchanged.
 
 **Return value — rewrite the tool's arguments:**
 
@@ -572,7 +572,7 @@ Shell hooks also accept the Claude Code-compatible format:
 
 Both formats are normalized internally to `{"action": "modify", "args": {...}}`.
 
-If a `pre_tool_call` callback exceeds `plugins.hook_callback_timeout` (or is still running from a previous timed-out fire), Hermes **fails closed**: the tool is blocked with a timeout message rather than proceeding without a policy decision.
+If a `pre_tool_call` callback exceeds `plugins.hook_callback_timeout` (or is still running from a previous timed-out fire), Hermes **fails closed**: the tool is blocked with a timeout message rather than proceeding without a policy decision. The same applies to a callback that raises: the block message names the callback and the error. A hung callback is skipped for a 60s suppression window; after that a new tool call runs it again (up to three abandoned workers per callback, so a permanently hung plugin blocks tool calls with a warning naming it instead of silently wedging the agent until restart).
 
 **Use cases:** Logging, audit trails, tool call counters, blocking dangerous operations, rate limiting, per-user policy enforcement, argument sanitization, path rewriting, injecting default parameters.
 
@@ -1214,6 +1214,8 @@ def my_callback(event, gateway, session_store, **kwargs):
 
 **Return value:** `None` or a dict. The first recognized action dict wins; remaining plugin results are ignored. Exceptions in plugin callbacks are caught and logged; the gateway always falls through to normal dispatch on error.
 
+Callbacks may be `async def`: they are awaited on the gateway's own event loop, so awaiting loop-bound work (an `asyncio.Event`, an aiohttp session, `asyncio.to_thread`) makes progress and other inbound messages keep flowing while the callback runs. The hook is intentionally not bounded by `plugins.hook_callback_timeout` — dropping or passing a message on timeout are both wrong for a policy gate — so a callback that never returns holds up dispatch of that message.
+
 | Return | Effect |
 |--------|--------|
 | `{"action": "skip", "reason": "..."}` | Drop the message — no agent reply, no pairing flow, no auth. Plugin is assumed to have handled it (e.g. silent-ingested into the transcript). |
@@ -1729,6 +1731,10 @@ profile's `HERMES_HOME`. `tool_name` and `tool_input` are `null` for non-tool ev
 {"action": "modify", "args": {"new_string": "fixed content"}}         // Hermes-canonical
 {"decision": "modify", "tool_input": {"new_string": "fixed content"}} // Claude-Code style
 
+// Escalate a pre_tool_call to the human-approval gate (Hermes-only; `message` and `rule_key`
+// are optional). Claude-Code's `{"decision": "approve"}` means auto-allow and is NOT mapped here:
+{"action": "approve", "message": "Why approval is required", "rule_key": "optional:scope"}
+
 // Inject context for pre_llm_call:
 {"context": "Today is Friday, 2026-04-17"}
 
@@ -1919,7 +1925,7 @@ Shell hooks run with **your full user credentials** — same trust boundary as a
 
 ### Ordering and precedence
 
-Both Python plugin hooks and shell hooks flow through the same `invoke_hook()` dispatcher. Python plugins are registered first (`discover_and_load()`), shell hooks second (`register_from_config()`), so Python `pre_tool_call` block decisions take precedence in tie cases. The first valid block wins — the aggregator returns as soon as any callback produces `{"action": "block", "message": str}` with a non-empty message.
+Both Python plugin hooks and shell hooks flow through the same `invoke_hook()` dispatcher. Python plugins are registered first (`discover_and_load()`), shell hooks second (`register_from_config()`), so Python `pre_tool_call` decisions take precedence in tie cases. The first valid block wins — the aggregator returns as soon as any callback produces `{"action": "block", "message": str}` with a non-empty message — and a block anywhere in the list outranks an `approve` returned earlier.
 
 ## Outbound Webhooks