fix(plugins): bound lifecycle hook execution on hot paths (#10048)

hermes_cli/plugins.py

@@ -34,13 +34,15 @@
 from __future__ import annotations
 
 import asyncio
+import contextvars
 import importlib.metadata
 import importlib.util
 import inspect
 import logging
 import os
 import sys
 import threading
+import time
 import types
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -51,6 +53,9 @@
 from hermes_cli.config import cfg_get
 from hermes_cli.middleware import OBSERVER_SCHEMA_VERSION, VALID_MIDDLEWARE
 
+_DEFAULT_HOOK_TIMEOUT_SECONDS = 2.0
+_HOOK_TIMEOUT_SUPPRESSION_SECONDS = 60.0
+
 
 def get_bundled_plugins_dir() -> Path:
     """Locate the bundled ``plugins/`` directory.
@@ -214,6 +219,24 @@ def _install_plugin_debug_handler(force: bool = False) -> None:
     "kanban_task_blocked",
 }
 
+_HOOK_TIMEOUT_BOUNDED_HOOKS: Set[str] = {
+    "post_tool_call",
+    "transform_terminal_output",
+    "transform_tool_result",
+    "transform_llm_output",
+    "pre_llm_call",
+    "post_llm_call",
+    "pre_api_request",
+    "post_api_request",
+    "api_request_error",
+    "pre_verify",
+    "on_session_start",
+    "on_session_end",
+}
+
+_HOOK_TIMEOUT_FAIL_CLOSED_HOOKS: Set[str] = {"pre_tool_call"}
+_HOOK_CALLER_THREAD_HOOKS: Set[str] = {"subagent_stop"}
+
 ENTRY_POINTS_GROUP = "hermes_agent.plugins"
 
 _NS_PARENT = "hermes_plugins"
@@ -270,6 +293,28 @@ def _get_enabled_plugins() -> Optional[set]:
         return None
 
 
+def _get_hook_timeout_seconds() -> float:
+    """Return the per-callback lifecycle hook deadline in seconds.
+
+    The value is configurable as ``plugins.hook_timeout_seconds`` and defaults
+    to a small fail-open deadline so hot-path lifecycle hooks cannot wedge a
+    user turn indefinitely.
+    """
+    try:
+        from hermes_cli.config import load_config
+        config = load_config()
+        raw_value = cfg_get(
+            config,
+            "plugins",
+            "hook_timeout_seconds",
+            default=_DEFAULT_HOOK_TIMEOUT_SECONDS,
+        )
+        value = float(raw_value)
+        return value if value > 0 else _DEFAULT_HOOK_TIMEOUT_SECONDS
+    except Exception:
+        return _DEFAULT_HOOK_TIMEOUT_SECONDS
+
+
 # ---------------------------------------------------------------------------
 # Data classes
 # ---------------------------------------------------------------------------
@@ -1271,6 +1316,11 @@ def __init__(self) -> None:
         # ``re.Pattern``, or a constraint dict); ``callback`` is an async
         # function with the slack_bolt signature ``(ack, body, action)``.
         self._slack_action_handlers: List[tuple] = []
+        self._hook_timeout_seconds = _get_hook_timeout_seconds()
+        self._hook_timeout_suppression_seconds = _HOOK_TIMEOUT_SUPPRESSION_SECONDS
+        self._hook_timeout_suppressed_until: Dict[tuple, float] = {}
+        self._hook_running_callbacks: Dict[tuple, object] = {}
+        self._hook_timeout_lock = threading.Lock()
 
     # -----------------------------------------------------------------------
     # Public
@@ -1912,16 +1962,84 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]:
         kwargs.setdefault("telemetry_schema_version", OBSERVER_SCHEMA_VERSION)
         callbacks = self._hooks.get(hook_name, [])
         results: List[Any] = []
+        use_timeout = (
+            hook_name in _HOOK_TIMEOUT_BOUNDED_HOOKS
+            or hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS
+        )
         for cb in callbacks:
+            callback_name = getattr(cb, "__name__", repr(cb))
+            callback_key = (hook_name, id(cb))
+            now = time.monotonic()
             try:
-                ret = cb(**kwargs)
+                if use_timeout:
+                    token = object()
+                    with self._hook_timeout_lock:
+                        suppressed_until = self._hook_timeout_suppressed_until.get(callback_key)
+                        running = callback_key in self._hook_running_callbacks
+                        if (suppressed_until is not None and suppressed_until > now) or running:
+                            logger.warning(
+                                "Hook '%s' callback %s skipped after previous timeout or while still running",
+                                hook_name,
+                                callback_name,
+                            )
+                            if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS:
+                                results.append({
+                                    "action": "block",
+                                    "message": "pre_tool_call plugin callback timed out or is still running",
+                                })
+                            continue
+                        if suppressed_until is not None:
+                            self._hook_timeout_suppressed_until.pop(callback_key, None)
+                        self._hook_running_callbacks[callback_key] = token
+                    context = contextvars.copy_context()
+                    done = threading.Event()
+                    result_holder: Dict[str, Any] = {}
+
+                    def run_callback() -> None:
+                        try:
+                            result_holder["result"] = context.run(cb, **kwargs)
+                        except Exception as exc:
+                            result_holder["exception"] = exc
+                        finally:
+                            with self._hook_timeout_lock:
+                                if self._hook_running_callbacks.get(callback_key) is token:
+                                    self._hook_running_callbacks.pop(callback_key, None)
+                            done.set()
+
+                    threading.Thread(
+                        target=run_callback,
+                        name="hermes-plugin-hook",
+                        daemon=True,
+                    ).start()
+                    if not done.wait(self._hook_timeout_seconds):
+                        with self._hook_timeout_lock:
+                            self._hook_timeout_suppressed_until[callback_key] = (
+                                time.monotonic() + self._hook_timeout_suppression_seconds
+                            )
+                        logger.warning(
+                            "Hook '%s' callback %s timed out after %.2fs",
+                            hook_name,
+                            callback_name,
+                            self._hook_timeout_seconds,
+                        )
+                        if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS:
+                            results.append({
+                                "action": "block",
+                                "message": "pre_tool_call plugin callback timed out or is still running",
+                            })
+                        continue
+                    if "exception" in result_holder:
+                        raise result_holder["exception"]
+                    ret = result_holder.get("result")
+                else:
+                    ret = cb(**kwargs)
                 if ret is not None:
                     results.append(ret)
             except Exception as exc:
                 logger.warning(
                     "Hook '%s' callback %s raised: %s",
                     hook_name,
-                    getattr(cb, "__name__", repr(cb)),
+                    callback_name,
                     exc,
                 )
         return results

tests/agent/test_system_prompt_restore.py

@@ -21,6 +21,7 @@
 import pytest
 
 from agent.conversation_loop import _restore_or_build_system_prompt
+from hermes_cli.plugins import PluginManager
 
 
 def _make_agent(session_db=None, prebuilt_prompt: str = "BUILT_PROMPT"):
@@ -139,6 +140,24 @@ def test_no_db_skips_persistence(self):
         agent._build_system_prompt.assert_called_once()
         assert agent._cached_system_prompt == "BUILT_PROMPT"
 
+    def test_slow_session_start_hook_does_not_delay_first_turn(self, monkeypatch):
+        manager = PluginManager()
+        manager._hook_timeout_seconds = 0.02
+
+        def slow_session_start(**kwargs):
+            import time
+            time.sleep(0.30)
+
+        manager._hooks["on_session_start"] = [slow_session_start]
+        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)
+        db = MagicMock()
+        agent = _make_agent(session_db=db)
+
+        _restore_or_build_system_prompt(agent, None, [])
+
+        assert agent._cached_system_prompt == "BUILT_PROMPT"
+        db.update_system_prompt.assert_called_once_with(agent.session_id, "BUILT_PROMPT")
+
 
 # ---------------------------------------------------------------------------
 # Silent-failure recovery — these are the new A/B logging paths