diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 2493d8f21edd..f63492ad4584 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -36,6 +36,7 @@ import asyncio import copy import hashlib +import contextvars import importlib.metadata import importlib.util import inspect @@ -46,6 +47,7 @@ import re import sys import threading +import time import types from contextlib import contextmanager from dataclasses import dataclass, field @@ -63,6 +65,7 @@ from utils import env_var_enabled, fast_safe_load from hermes_cli.config import cfg_get, load_config_readonly from hermes_cli.middleware import OBSERVER_SCHEMA_VERSION, VALID_MIDDLEWARE +from agent.deadline import run_bounded_sync from hermes_cli.plugin_capabilities import ( # noqa: F401 — re-exported CAPABILITY_REGISTRY, VALID_CAPABILITY_IDS, @@ -77,7 +80,6 @@ legacy_relay_plugin_keys, ) - def get_bundled_plugins_dir() -> Path: """Locate the bundled ``plugins/`` directory. @@ -396,6 +398,17 @@ def _install_plugin_debug_handler(force: bool = False) -> None: "transform_api_error_classification", } +_DEFAULT_HOOK_TIMEOUT_SECONDS = 2.0 +_HOOK_TIMEOUT_SUPPRESSION_SECONDS = 60.0 +_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" ENTRY_POINT_CAPABILITIES_GROUP = "hermes_agent.plugin_capabilities" @@ -618,6 +631,20 @@ def _get_enabled_plugins() -> Optional[set]: return None +def _get_hook_timeout_seconds() -> float: + """Return the per-callback lifecycle hook deadline in seconds.""" + try: + config = load_config_readonly() + 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 # --------------------------------------------------------------------------- @@ -3467,6 +3494,11 @@ def __init__(self, scope_key: Optional[str] = None) -> None: # full plugin loads. self._predeclared_modules: Dict[str, types.ModuleType] = {} self._predeclared_tools: Dict[str, List[str]] = {} + 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, threading.Event] = {} + self._hook_timeout_lock = threading.Lock() # ----------------------------------------------------------------------- # Registration ledger internals @@ -3665,9 +3697,31 @@ def _unload_scoped( ] found = bool(target_keys or registrations) + hook_names = {registration.key for registration in registrations if registration.kind == "hook"} self._dispose_registrations(registrations) self._forget_registrations(registrations) + stale_events = [] + if not unload_all and hook_names: + with self._hook_timeout_lock: + live_callback_keys = { + (hook_name, id(callback)) + for hook_name in hook_names + for callback in self._hooks.get(hook_name, []) + } + stale_keys = [ + key + for key in self._hook_running_callbacks + if key[0] in hook_names and key not in live_callback_keys + ] + for key in stale_keys: + event = self._hook_running_callbacks.pop(key, None) + if event is not None: + stale_events.append(event) + self._hook_timeout_suppressed_until.pop(key, None) + for event in stale_events: + event.set() + if unload_all: # The handles are authoritative for global registries, while the # manager-local containers are also reset to clear legacy/manual @@ -3719,6 +3773,10 @@ def _unload_scoped( self._ownership_ledger.clear() self._plugins.clear() self._hooks.clear() + with self._hook_timeout_lock: + self._hook_timeout_suppressed_until.clear() + stale_events.extend(self._hook_running_callbacks.values()) + self._hook_running_callbacks.clear() self._middleware.clear() self._plugin_tool_names.clear() self._plugin_platform_names.clear() @@ -3738,6 +3796,9 @@ def _unload_scoped( for key in target_keys: self._plugins.pop(key, None) + for event in stale_events: + event.set() + return found # ----------------------------------------------------------------------- @@ -5133,15 +5194,130 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: callbacks = self._hooks.get(hook_name, []) results: List[Any] = [] for cb in callbacks: + callback_name = getattr(cb, "__name__", repr(cb)) + callback_key = (hook_name, id(cb)) + bounded = hook_name in _HOOK_TIMEOUT_BOUNDED_HOOKS or hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS + if ( + hook_name in _HOOK_CALLER_THREAD_HOOKS + or not bounded + or callback_name.startswith("shell_hook[") + ): + try: + ret = self._invoke_hook_callback(cb, kwargs) + if ret is not None: + results.append(ret) + except Exception as exc: + logger.warning("Hook '%s' callback %s raised: %s", hook_name, callback_name, exc) + continue + skip_reason = None + while True: + now = time.monotonic() + with self._hook_timeout_lock: + callback_live = any( + registered is cb + for registered in self._hooks.get(hook_name, []) + ) + if not callback_live: + skip_reason = "unloaded" + running_event = None + else: + suppressed_until = self._hook_timeout_suppressed_until.get(callback_key) + if suppressed_until is not None and suppressed_until > now: + skip_reason = "suppressed" + running_event = None + else: + if suppressed_until is not None: + self._hook_timeout_suppressed_until.pop(callback_key, None) + running_event = self._hook_running_callbacks.get(callback_key) + if running_event is None: + running_event = threading.Event() + self._hook_running_callbacks[callback_key] = running_event + break + if skip_reason is not None: + break + if running_event.wait(self._hook_timeout_seconds): + continue + with self._hook_timeout_lock: + if self._hook_running_callbacks.get(callback_key) is running_event: + self._hook_timeout_suppressed_until[callback_key] = ( + time.monotonic() + self._hook_timeout_suppression_seconds + ) + skip_reason = "active callback timed out" + break + continue + if skip_reason is not None: + logger.warning( + "Hook '%s' callback %s skipped while %s; continuing %s", + hook_name, + callback_name, + skip_reason, + "fail-closed" if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS else "fail-open", + ) + if ( + hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS + and skip_reason != "unloaded" + ): + results.append({"action": "block", "message": "pre_tool_call plugin callback timed out or is still running"}) + continue + context = contextvars.copy_context() + + def bounded_callback( + callback=cb, + callback_key=callback_key, + context=context, + running_event=running_event, + ) -> Any: + try: + with self._hook_timeout_lock: + callback_live = any( + registered is callback + for registered in self._hooks.get(hook_name, []) + ) + state_is_current = ( + callback_live + and self._hook_running_callbacks.get(callback_key) is running_event + ) + if not state_is_current: + return None + return context.run(self._invoke_hook_callback, callback, kwargs) + finally: + with self._hook_timeout_lock: + if self._hook_running_callbacks.get(callback_key) is running_event: + self._hook_running_callbacks.pop(callback_key, None) + running_event.set() try: - ret = self._invoke_hook_callback(cb, kwargs) + bounded_result = run_bounded_sync( + bounded_callback, + self._hook_timeout_seconds, + label=f"plugin-hook:{hook_name}:{callback_name}", + ) + if bounded_result.timed_out: + with self._hook_timeout_lock: + callback_live = any( + registered is cb + for registered in self._hooks.get(hook_name, []) + ) + state_is_current = ( + callback_live + and self._hook_running_callbacks.get(callback_key) is running_event + ) + if state_is_current: + self._hook_timeout_suppressed_until[callback_key] = ( + time.monotonic() + self._hook_timeout_suppression_seconds + ) + if not state_is_current: + continue + 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 + ret = bounded_result.value 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 diff --git a/tests/agent/test_system_prompt_restore.py b/tests/agent/test_system_prompt_restore.py index ddbbc73c85b7..c7c4f57a7511 100644 --- a/tests/agent/test_system_prompt_restore.py +++ b/tests/agent/test_system_prompt_restore.py @@ -16,11 +16,14 @@ from __future__ import annotations import logging +import threading +import time from unittest.mock import MagicMock 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"): @@ -143,6 +146,29 @@ 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 + release = threading.Event() + + def slow_session_start(**kwargs): + release.wait(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) + + started = time.perf_counter() + try: + _restore_or_build_system_prompt(agent, None, []) + finally: + release.set() + + assert agent._cached_system_prompt == "BUILT_PROMPT" + db.update_system_prompt.assert_called_once_with(agent.session_id, "BUILT_PROMPT") + assert time.perf_counter() - started < 0.50 + # --------------------------------------------------------------------------- # Silent-failure recovery — these are the new A/B logging paths diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index b533cd7d4d0d..eee538d45a71 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -3,12 +3,16 @@ import logging import json import sys +import time +import threading +from contextvars import ContextVar import types from pathlib import Path from unittest.mock import MagicMock, patch import pytest import yaml +import hermes_cli.plugins as plugins from hermes_cli.plugins import ( ENTRY_POINTS_GROUP, @@ -781,7 +785,37 @@ def test_entrypoint_dotted_name_never_imports_parent_package( class TestPluginHooks: """Tests for lifecycle hook registration and invocation.""" + def test_hook_policies_are_disjoint_and_valid(self): + policies = ( + plugins._HOOK_TIMEOUT_BOUNDED_HOOKS, + plugins._HOOK_TIMEOUT_FAIL_CLOSED_HOOKS, + plugins._HOOK_CALLER_THREAD_HOOKS, + ) + assert all(policy <= VALID_HOOKS for policy in policies) + assert all( + left.isdisjoint(right) + for index, left in enumerate(policies) + for right in policies[index + 1:] + ) + assert {"pre_verify", "on_session_start", "on_session_end"} <= ( + plugins._HOOK_TIMEOUT_BOUNDED_HOOKS + ) + assert plugins._HOOK_TIMEOUT_FAIL_CLOSED_HOOKS == {"pre_tool_call"} + assert plugins._HOOK_CALLER_THREAD_HOOKS == {"subagent_stop"} + + def test_hook_timeout_config_uses_positive_values_only(self, monkeypatch): + configs = iter( + [ + {"plugins": {"hook_timeout_seconds": 1.5}}, + {"plugins": {"hook_timeout_seconds": 0}}, + {"plugins": {"hook_timeout_seconds": "invalid"}}, + ] + ) + monkeypatch.setattr(plugins, "load_config_readonly", lambda: next(configs)) + assert plugins._get_hook_timeout_seconds() == 1.5 + assert plugins._get_hook_timeout_seconds() == plugins._DEFAULT_HOOK_TIMEOUT_SECONDS + assert plugins._get_hook_timeout_seconds() == plugins._DEFAULT_HOOK_TIMEOUT_SECONDS def test_pre_gateway_dispatch_collects_action_dicts(self, tmp_path, monkeypatch): """pre_gateway_dispatch callbacks return action dicts (skip/rewrite/allow).""" @@ -810,6 +844,435 @@ def test_pre_gateway_dispatch_collects_action_dicts(self, tmp_path, monkeypatch) + def test_invoke_hook_adds_observer_schema_version(self, tmp_path, monkeypatch): + """invoke_hook() supplies the observer schema version for all hooks.""" + plugins_dir = tmp_path / "hermes_test" / "plugins" + _make_plugin_dir( + plugins_dir, + "schema_plugin", + register_body=( + 'ctx.register_hook("pre_tool_call", ' + 'lambda **kw: kw.get("telemetry_schema_version"))' + ), + ) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) + + mgr = PluginManager() + mgr.discover_and_load() + + assert mgr.invoke_hook("pre_tool_call", tool_name="test", args={}) == [ + "hermes.observer.v1" + ] + + def test_hook_exception_does_not_propagate(self, tmp_path, monkeypatch): + """A hook callback that raises does NOT crash the caller.""" + plugins_dir = tmp_path / "hermes_test" / "plugins" + _make_plugin_dir( + plugins_dir, "bad_hook", + register_body='ctx.register_hook("post_tool_call", lambda **kw: 1/0)', + ) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) + + mgr = PluginManager() + mgr.discover_and_load() + + # Should not raise despite 1/0 + mgr.invoke_hook("post_tool_call", tool_name="x", args={}, result="r", task_id="") + + def test_slow_hook_times_out_and_later_callbacks_still_run(self, caplog): + """A hung callback is skipped after the deadline and dispatch continues.""" + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.02 + calls = [] + release = threading.Event() + + def before(**kwargs): + calls.append("before") + return "before-result" + + def slow(**kwargs): + calls.append("slow-start") + release.wait(30) + calls.append("slow-end") + return "slow-result" + + def after(**kwargs): + calls.append("after") + return "after-result" + + mgr._hooks["pre_llm_call"] = [before, slow, after] + + start = time.perf_counter() + with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): + results = mgr.invoke_hook( + "pre_llm_call", + session_id="s1", + user_message="hi", + conversation_history=[], + is_first_turn=True, + model="test", + ) + elapsed = time.perf_counter() - start + + try: + assert elapsed < 0.20 + assert results == ["before-result", "after-result"] + assert calls[:3] == ["before", "slow-start", "after"] + assert any("timed out" in r.getMessage() for r in caplog.records) + finally: + release.set() + + def test_distinct_hung_callbacks_do_not_starve_healthy_hooks(self, caplog): + """Each timed-out callback has isolated capacity.""" + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.02 + mgr._hook_timeout_suppression_seconds = 30.0 + calls = [] + release = threading.Event() + + def slow(index): + def callback(**kwargs): + calls.append((index, "start")) + release.wait(30) + calls.append((index, "end")) + return callback + + def healthy(**kwargs): + calls.append(("healthy", "run")) + return "healthy-result" + + slow_callbacks = [slow(index) for index in range(4)] + mgr._hooks["pre_llm_call"] = [*slow_callbacks, healthy] + + with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): + started = time.perf_counter() + results = mgr.invoke_hook("pre_llm_call", session_id="s1") + elapsed = time.perf_counter() - started + + try: + assert elapsed < 0.20 + assert results == ["healthy-result"] + assert calls.count(("healthy", "run")) == 1 + assert all(calls.count((index, "start")) == 1 for index in range(4)) + assert any("timed out" in record.getMessage() for record in caplog.records) + finally: + release.set() + + @pytest.mark.parametrize("hook_name", ["on_session_start", "on_session_end"]) + def test_session_hooks_are_bounded(self, hook_name, caplog): + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.02 + release = threading.Event() + + def slow(**kwargs): + release.wait(30) + mgr._hooks[hook_name] = [slow] + + try: + with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): + started = time.perf_counter() + assert mgr.invoke_hook(hook_name, session_id="s1") == [] + elapsed = time.perf_counter() - started + assert elapsed < 0.20 + assert any(hook_name in record.getMessage() for record in caplog.records) + finally: + release.set() + + def test_pre_tool_call_timeout_and_suppression_block(self, monkeypatch): + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.02 + release = threading.Event() + started = [] + + def slow(**kwargs): + started.append(True) + release.wait(30) + + mgr._hooks["pre_tool_call"] = [slow] + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: mgr) + + try: + first = get_pre_tool_call_block_message("terminal", {}) + second = get_pre_tool_call_block_message("terminal", {}) + assert first and "timed out" in first + assert second and "timed out" in second + assert started == [True] + finally: + release.set() + + def test_concurrent_healthy_callbacks_wait_without_blocking(self): + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.2 + entered = threading.Event() + release = threading.Event() + second_started = threading.Event() + results = [] + errors = [] + + def healthy(**kwargs): + entered.set() + release.wait(1) + + def invoke(): + try: + results.append(mgr.invoke_hook("pre_tool_call", tool_name="terminal")) + except BaseException as exc: + errors.append(exc) + + mgr._hooks["pre_tool_call"] = [healthy] + first = threading.Thread(target=invoke) + second = threading.Thread( + target=lambda: (second_started.set(), invoke()), + ) + first.start() + assert entered.wait(1) + second.start() + assert second_started.wait(1) + time.sleep(0.05) + release.set() + first.join(1) + second.join(1) + + assert errors == [] + assert results == [[], []] + + def test_shell_hook_callback_keeps_its_caller_thread_and_timeout(self): + mgr = PluginManager() + entered = threading.Event() + release = threading.Event() + callback_thread = [] + caller_thread = [] + + def shell_hook(**kwargs): + callback_thread.append(threading.current_thread()) + entered.set() + release.wait(1) + return {"action": "allow"} + + shell_hook.__name__ = "shell_hook[pre_tool_call:echo]" + mgr._hooks["pre_tool_call"] = [shell_hook] + + def invoke(): + caller_thread.append(threading.current_thread()) + return mgr.invoke_hook("pre_tool_call", tool_name="terminal") + + thread = threading.Thread(target=invoke) + thread.start() + assert entered.wait(1) + release.set() + thread.join(1) + + assert callback_thread == caller_thread + + + def test_bounded_hook_uses_shared_adapter_and_caller_context(self): + mgr = PluginManager() + marker = ContextVar("plugin-test-marker", default="missing") + marker.set("caller-context") + + def narrow(session_id, telemetry_schema_version): + return marker.get(), session_id, telemetry_schema_version + + mgr._hooks["pre_llm_call"] = [narrow] + with patch("hermes_cli.plugins.run_bounded_sync", wraps=plugins.run_bounded_sync) as bounded: + assert mgr.invoke_hook("pre_llm_call", session_id="session-1", extra="ignored") == [ + ("caller-context", "session-1", "hermes.observer.v1") + ] + + bounded.assert_called_once() + assert bounded.call_args.args[1] == mgr._hook_timeout_seconds + assert bounded.call_args.kwargs["label"].startswith("plugin-hook:pre_llm_call:narrow") + + def test_bounded_callback_closes_over_registered_callback(self, monkeypatch): + mgr = PluginManager() + calls = [] + deferred = [] + + def first(**kwargs): + calls.append("first") + return "first-result" + + def second(**kwargs): + calls.append("second") + return "second-result" + + mgr._hooks["pre_llm_call"] = [first, second] + + def delayed_runner(callback, timeout, **kwargs): + deferred.append(callback) + if len(deferred) == 1: + return types.SimpleNamespace(timed_out=True, value=None) + return types.SimpleNamespace(timed_out=False, value=callback()) + + monkeypatch.setattr(plugins, "run_bounded_sync", delayed_runner) + + assert mgr.invoke_hook("pre_llm_call", session_id="session-1") == [ + "second-result" + ] + deferred[0]() + assert calls == ["second", "first"] + + def test_hook_exception_still_allows_later_return_values(self, caplog): + """Bounded dispatch keeps the prior fail-open exception behavior.""" + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.1 + + def before(**kwargs): + return "before-result" + + def broken(**kwargs): + raise RuntimeError("boom") + + def after(**kwargs): + return "after-result" + + mgr._hooks["pre_llm_call"] = [before, broken, after] + + with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): + results = mgr.invoke_hook( + "pre_llm_call", + session_id="s1", + user_message="hi", + conversation_history=[], + is_first_turn=True, + model="test", + ) + + assert results == ["before-result", "after-result"] + assert any("raised: boom" in r.getMessage() for r in caplog.records) + + def test_unload_clears_hook_timeout_state(self): + mgr = PluginManager() + callback_key = ("pre_tool_call", 1) + mgr._hook_timeout_suppressed_until[callback_key] = time.monotonic() + 60 + event = threading.Event() + mgr._hook_running_callbacks[callback_key] = event + + mgr.unload() + + assert mgr._hook_timeout_suppressed_until == {} + assert mgr._hook_running_callbacks == {} + assert event.is_set() + + def test_scoped_unload_clears_removed_hook_timeout_state(self): + mgr = PluginManager() + callback = lambda **kwargs: None + PluginContext(PluginManifest(name="test-plugin"), mgr).register_hook( + "pre_tool_call", callback + ) + callback_key = ("pre_tool_call", id(callback)) + mgr._hook_timeout_suppressed_until[callback_key] = time.monotonic() + 60 + mgr._hook_running_callbacks[callback_key] = threading.Event() + + mgr.unload("test-plugin") + + assert mgr._hook_timeout_suppressed_until == {} + assert mgr._hook_running_callbacks == {} + + def test_scoped_unload_invalidates_waiting_callback_snapshot(self): + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.2 + entered = threading.Event() + release = threading.Event() + calls = [] + + def callback(**kwargs): + calls.append(True) + entered.set() + release.wait(1) + + PluginContext(PluginManifest(name="test-plugin"), mgr).register_hook( + "pre_tool_call", callback + ) + results = [] + first = threading.Thread( + target=lambda: results.append(mgr.invoke_hook("pre_tool_call")) + ) + second = threading.Thread( + target=lambda: results.append(mgr.invoke_hook("pre_tool_call")) + ) + first.start() + assert entered.wait(1) + second.start() + time.sleep(0.05) + mgr.unload("test-plugin") + release.set() + first.join(1) + second.join(1) + + assert calls == [True] + assert results == [[], []] + + def test_unload_prevents_timeout_state_repopulation(self, monkeypatch): + mgr = PluginManager() + PluginContext(PluginManifest(name="test-plugin"), mgr).register_hook( + "pre_tool_call", lambda **kwargs: None + ) + + def timed_out_runner(callback, timeout, **kwargs): + mgr.unload("test-plugin") + return types.SimpleNamespace(timed_out=True, value=None) + + monkeypatch.setattr(plugins, "run_bounded_sync", timed_out_runner) + + assert mgr.invoke_hook("pre_tool_call") == [] + assert mgr._hook_timeout_suppressed_until == {} + assert mgr._hook_running_callbacks == {} + + def test_non_hot_hook_stays_on_caller_thread(self): + """Lifecycle hooks preserve caller-thread semantics outside hot paths.""" + mgr = PluginManager() + seen_threads = [] + caller_thread = threading.current_thread() + + def capture(**kwargs): + seen_threads.append(threading.current_thread()) + + mgr._hooks["subagent_stop"] = [capture] + + mgr.invoke_hook( + "subagent_stop", + parent_session_id="parent-1", + child_status="completed", + ) + + assert seen_threads == [caller_thread] + + def test_hook_return_values_collected(self, tmp_path, monkeypatch): + """invoke_hook() collects non-None return values from callbacks.""" + plugins_dir = tmp_path / "hermes_test" / "plugins" + _make_plugin_dir( + plugins_dir, "ctx_plugin", + register_body=( + 'ctx.register_hook("pre_llm_call", ' + 'lambda **kw: {"context": "memory from plugin"})' + ), + ) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) + + mgr = PluginManager() + mgr.discover_and_load() + + results = mgr.invoke_hook("pre_llm_call", session_id="s1", user_message="hi", + conversation_history=[], is_first_turn=True, model="test") + assert len(results) == 1 + assert results[0] == {"context": "memory from plugin"} + + def test_hook_none_returns_excluded(self, tmp_path, monkeypatch): + """invoke_hook() excludes None returns from the result list.""" + plugins_dir = tmp_path / "hermes_test" / "plugins" + _make_plugin_dir( + plugins_dir, "none_hook", + register_body='ctx.register_hook("post_llm_call", lambda **kw: None)', + ) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test")) + + mgr = PluginManager() + mgr.discover_and_load() + + results = mgr.invoke_hook("post_llm_call", session_id="s1", + user_message="hi", assistant_response="bye", model="test") + assert results == [] def test_request_hooks_are_invokeable(self, tmp_path, monkeypatch): plugins_dir = tmp_path / "hermes_test" / "plugins" diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index a72da553be4f..fae1bb6da929 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -2213,8 +2213,39 @@ def test_sequential_blocked_tool_skips_checkpoints_and_callbacks(self, agent, mo assert messages[0]["role"] == "tool" assert json.loads(messages[0]["content"]) == {"error": "Blocked by policy"} + def test_timed_out_pre_tool_hook_blocks_before_tool_dispatch(self, agent, monkeypatch): + from hermes_cli.plugins import PluginManager + manager = PluginManager() + manager._hook_timeout_seconds = 0.02 + release = threading.Event() + def slow_pre_tool_call(**kwargs): + release.wait(30) + + manager._hooks["pre_tool_call"] = [slow_pre_tool_call] + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager) + + tool_call = _mock_tool_call( + name="write_file", + arguments='{"path":"test.txt","content":"hello"}', + call_id="timeout-1", + ) + mock_msg = _mock_assistant_msg(content="", tool_calls=[tool_call]) + messages = [] + agent._checkpoint_mgr.enabled = True + agent._checkpoint_mgr.ensure_checkpoint = MagicMock( + side_effect=AssertionError("checkpoint should not run") + ) + + try: + with patch("run_agent.handle_function_call", side_effect=AssertionError("should not run")): + agent._execute_tool_calls_sequential(mock_msg, messages, "task-1") + + agent._checkpoint_mgr.ensure_checkpoint.assert_not_called() + assert json.loads(messages[0]["content"])["error"] + finally: + release.set() @pytest.mark.parametrize("concurrent", [False, True]) def test_tool_execution_middleware_replacement_emits_one_terminal_hook(