From 9a258a00d7f0ca4e58a0ca434acb74465af79e66 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 5 Jun 2026 06:25:10 -0400 Subject: [PATCH 01/10] fix(plugins): bound lifecycle hook execution on hot paths (#10048) --- hermes_cli/plugins.py | 87 ++++++++++++++- tests/hermes_cli/test_plugins.py | 183 +++++++++++++++++++++++++++++++ 2 files changed, 267 insertions(+), 3 deletions(-) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 2493d8f21edd..12caa5d9b3de 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: Set[tuple] = set() + self._hook_timeout_lock = threading.Lock() # ----------------------------------------------------------------------- # Registration ledger internals @@ -5133,15 +5165,64 @@ 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: + 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 + now = time.monotonic() + with self._hook_timeout_lock: + suppressed_until = self._hook_timeout_suppressed_until.get(callback_key) + active = callback_key in self._hook_running_callbacks + if active or (suppressed_until is not None and suppressed_until > now): + logger.warning( + "Hook '%s' callback %s skipped while active or suppressed; continuing %s", + hook_name, + callback_name, + "fail-closed" if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS else "fail-open", + ) + if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS: + results.append({"action": "block", "message": f"BLOCKED: plugin hook '{callback_name}' is unavailable"}) + continue + if suppressed_until is not None: + self._hook_timeout_suppressed_until.pop(callback_key, None) + self._hook_running_callbacks.add(callback_key) + context = contextvars.copy_context() + + def bounded_callback() -> Any: + try: + return context.run(self._invoke_hook_callback, cb, kwargs) + finally: + with self._hook_timeout_lock: + self._hook_running_callbacks.discard(callback_key) 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: + self._hook_timeout_suppressed_until[callback_key] = ( + time.monotonic() + self._hook_timeout_suppression_seconds + ) + if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS: + results.append({"action": "block", "message": f"BLOCKED: plugin hook '{callback_name}' timed out"}) + 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/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index b533cd7d4d0d..fdba09eb4734 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -3,6 +3,7 @@ import logging import json import sys +import time import types from pathlib import Path from unittest.mock import MagicMock, patch @@ -810,6 +811,188 @@ 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 = [] + + def before(**kwargs): + calls.append("before") + return "before-result" + + def slow(**kwargs): + calls.append("slow-start") + time.sleep(0.25) + 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 + + 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) + + def test_timed_out_hook_is_suppressed_without_blocking_healthy_hooks(self, caplog): + """A repeatedly hung callback does not consume the shared executor forever.""" + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.02 + mgr._hook_timeout_suppression_seconds = 30.0 + calls = [] + + def slow(**kwargs): + calls.append("slow-start") + time.sleep(0.30) + calls.append("slow-end") + return "slow-result" + + def healthy(**kwargs): + calls.append("healthy") + return "healthy-result" + + mgr._hooks["pre_llm_call"] = [slow, healthy] + + with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"): + first = mgr.invoke_hook( + "pre_llm_call", + session_id="s1", + user_message="hi", + conversation_history=[], + is_first_turn=True, + model="test", + ) + second = mgr.invoke_hook( + "pre_llm_call", + session_id="s1", + user_message="hi", + conversation_history=[], + is_first_turn=True, + model="test", + ) + + assert first == ["healthy-result"] + assert second == ["healthy-result"] + assert calls.count("slow-start") == 1 + assert calls.count("healthy") == 2 + assert any("timed out" in r.getMessage() for r in caplog.records) + assert any("skipped after previous timeout" in r.getMessage() for r in caplog.records) + + 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_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" From 10ca98d47ef2f379951974af17fbfccfca292258 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 12 Jun 2026 07:51:47 -0400 Subject: [PATCH 02/10] fix(plugins): preserve lifecycle hook thread affinity --- tests/hermes_cli/test_plugins.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index fdba09eb4734..9c7685a70544 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -4,6 +4,7 @@ import json import sys import time +import threading import types from pathlib import Path from unittest.mock import MagicMock, patch @@ -958,6 +959,25 @@ def after(**kwargs): assert results == ["before-result", "after-result"] assert any("raised: boom" in r.getMessage() for r in caplog.records) + 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" From 5f3392b864481e084c870cf38d978331a35bcf04 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 13 Jul 2026 23:22:21 -0400 Subject: [PATCH 03/10] fix(plugins): fail closed when timed hooks gate tools (#10048) --- tests/agent/test_system_prompt_restore.py | 19 ++++ tests/hermes_cli/test_plugins.py | 100 +++++++++++++++------- tests/run_agent/test_run_agent.py | 28 ++++++ 3 files changed, 116 insertions(+), 31 deletions(-) diff --git a/tests/agent/test_system_prompt_restore.py b/tests/agent/test_system_prompt_restore.py index ddbbc73c85b7..95ee27b66415 100644 --- a/tests/agent/test_system_prompt_restore.py +++ b/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"): @@ -143,6 +144,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 diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 9c7685a70544..17fdd9d926df 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -11,6 +11,7 @@ import pytest import yaml +import hermes_cli.plugins as plugins from hermes_cli.plugins import ( ENTRY_POINTS_GROUP, @@ -781,6 +782,24 @@ def test_entrypoint_dotted_name_never_imports_parent_package( class TestPluginHooks: + 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"} + """Tests for lifecycle hook registration and invocation.""" @@ -886,49 +905,68 @@ def after(**kwargs): assert calls[:3] == ["before", "slow-start", "after"] assert any("timed out" in r.getMessage() for r in caplog.records) - def test_timed_out_hook_is_suppressed_without_blocking_healthy_hooks(self, caplog): - """A repeatedly hung callback does not consume the shared executor forever.""" + 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 = [] - def slow(**kwargs): - calls.append("slow-start") - time.sleep(0.30) - calls.append("slow-end") - return "slow-result" + def slow(index): + def callback(**kwargs): + calls.append((index, "start")) + time.sleep(0.30) + calls.append((index, "end")) + return callback def healthy(**kwargs): - calls.append("healthy") + calls.append(("healthy", "run")) return "healthy-result" - mgr._hooks["pre_llm_call"] = [slow, healthy] + 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"): - first = mgr.invoke_hook( - "pre_llm_call", - session_id="s1", - user_message="hi", - conversation_history=[], - is_first_turn=True, - model="test", - ) - second = mgr.invoke_hook( - "pre_llm_call", - session_id="s1", - user_message="hi", - conversation_history=[], - is_first_turn=True, - model="test", - ) + started = time.perf_counter() + results = mgr.invoke_hook("pre_llm_call", session_id="s1") + elapsed = time.perf_counter() - started - assert first == ["healthy-result"] - assert second == ["healthy-result"] - assert calls.count("slow-start") == 1 - assert calls.count("healthy") == 2 - assert any("timed out" in r.getMessage() for r in caplog.records) - assert any("skipped after previous timeout" in r.getMessage() for r in caplog.records) + 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) + + @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 + + def slow(**kwargs): + time.sleep(0.30) + mgr._hooks[hook_name] = [slow] + + 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) + + def test_pre_tool_call_timeout_and_suppression_block(self, monkeypatch): + mgr = PluginManager() + mgr._hook_timeout_seconds = 0.02 + + def slow(**kwargs): + time.sleep(0.30) + + mgr._hooks["pre_tool_call"] = [slow] + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: mgr) + + 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 def test_hook_exception_still_allows_later_return_values(self, caplog): """Bounded dispatch keeps the prior fail-open exception behavior.""" diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index a72da553be4f..9e82f368287d 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -2213,8 +2213,36 @@ 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 + def slow_pre_tool_call(**kwargs): + import time + time.sleep(0.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") + ) + + 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"] @pytest.mark.parametrize("concurrent", [False, True]) def test_tool_execution_middleware_replacement_emits_one_terminal_hook( From d2b7c2e405dee2f5b3719637585f5d1af1ffdb18 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 20 Aug 2026 04:24:17 -0400 Subject: [PATCH 04/10] fix(plugins): preserve bounded callback ownership Keep abandoned callback workers keyed to their original callback and retain the fail-closed pre_tool_call directive while using the shared deadline primitive. --- hermes_cli/plugins.py | 6 +-- tests/hermes_cli/test_plugins.py | 81 +++++++++++++++++++++++--------- 2 files changed, 61 insertions(+), 26 deletions(-) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 12caa5d9b3de..276825e5feac 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -5188,14 +5188,14 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: "fail-closed" if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS else "fail-open", ) if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS: - results.append({"action": "block", "message": f"BLOCKED: plugin hook '{callback_name}' is unavailable"}) + 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.add(callback_key) context = contextvars.copy_context() - def bounded_callback() -> Any: + def bounded_callback(callback=cb, callback_key=callback_key, context=context) -> Any: try: return context.run(self._invoke_hook_callback, cb, kwargs) finally: @@ -5213,7 +5213,7 @@ def bounded_callback() -> Any: time.monotonic() + self._hook_timeout_suppression_seconds ) if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS: - results.append({"action": "block", "message": f"BLOCKED: plugin hook '{callback_name}' timed out"}) + 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: diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 17fdd9d926df..3d88d255ce8b 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -5,6 +5,7 @@ import sys import time import threading +from contextvars import ContextVar import types from pathlib import Path from unittest.mock import MagicMock, patch @@ -871,6 +872,7 @@ def test_slow_hook_times_out_and_later_callbacks_still_run(self, caplog): mgr = PluginManager() mgr._hook_timeout_seconds = 0.02 calls = [] + release = threading.Event() def before(**kwargs): calls.append("before") @@ -878,7 +880,7 @@ def before(**kwargs): def slow(**kwargs): calls.append("slow-start") - time.sleep(0.25) + release.wait(30) calls.append("slow-end") return "slow-result" @@ -900,10 +902,13 @@ def after(**kwargs): ) elapsed = time.perf_counter() - start - 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) + 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.""" @@ -911,11 +916,12 @@ def test_distinct_hung_callbacks_do_not_starve_healthy_hooks(self, caplog): 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")) - time.sleep(0.30) + release.wait(30) calls.append((index, "end")) return callback @@ -931,42 +937,71 @@ def healthy(**kwargs): results = mgr.invoke_hook("pre_llm_call", session_id="s1") elapsed = time.perf_counter() - started - 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) + 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): - time.sleep(0.30) + release.wait(30) mgr._hooks[hook_name] = [slow] - 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) + 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() def slow(**kwargs): - time.sleep(0.30) + release.wait(30) mgr._hooks["pre_tool_call"] = [slow] monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: mgr) - 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 + 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 + finally: + release.set() + + 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_hook_exception_still_allows_later_return_values(self, caplog): """Bounded dispatch keeps the prior fail-open exception behavior.""" From e3804defe6cae75e64ed06d39e5003d6c1daaef4 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 20 Aug 2026 04:40:26 -0400 Subject: [PATCH 05/10] fix(plugins): preserve callback identity and reset timeout state --- hermes_cli/plugins.py | 5 ++- tests/agent/test_system_prompt_restore.py | 10 +++-- tests/hermes_cli/test_plugins.py | 54 +++++++++++++++++++++++ tests/run_agent/test_run_agent.py | 15 ++++--- 4 files changed, 74 insertions(+), 10 deletions(-) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 276825e5feac..a04bbf9c2584 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -3751,6 +3751,9 @@ def _unload_scoped( self._ownership_ledger.clear() self._plugins.clear() self._hooks.clear() + with self._hook_timeout_lock: + self._hook_timeout_suppressed_until.clear() + self._hook_running_callbacks.clear() self._middleware.clear() self._plugin_tool_names.clear() self._plugin_platform_names.clear() @@ -5197,7 +5200,7 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: def bounded_callback(callback=cb, callback_key=callback_key, context=context) -> Any: try: - return context.run(self._invoke_hook_callback, cb, kwargs) + return context.run(self._invoke_hook_callback, callback, kwargs) finally: with self._hook_timeout_lock: self._hook_running_callbacks.discard(callback_key) diff --git a/tests/agent/test_system_prompt_restore.py b/tests/agent/test_system_prompt_restore.py index 95ee27b66415..cbbf8d5be091 100644 --- a/tests/agent/test_system_prompt_restore.py +++ b/tests/agent/test_system_prompt_restore.py @@ -16,6 +16,7 @@ from __future__ import annotations import logging +import threading from unittest.mock import MagicMock import pytest @@ -147,17 +148,20 @@ def test_no_db_skips_persistence(self): 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): - import time - time.sleep(0.30) + 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) - _restore_or_build_system_prompt(agent, None, []) + 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") diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 3d88d255ce8b..9e133e57fbac 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -801,6 +801,20 @@ def test_hook_policies_are_disjoint_and_valid(self): 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 + """Tests for lifecycle hook registration and invocation.""" @@ -1003,6 +1017,35 @@ def narrow(session_id, telemetry_schema_version): 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() @@ -1032,6 +1075,17 @@ def after(**kwargs): 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 + mgr._hook_running_callbacks.add(callback_key) + + mgr.unload() + + assert mgr._hook_timeout_suppressed_until == {} + assert mgr._hook_running_callbacks == set() + def test_non_hot_hook_stays_on_caller_thread(self): """Lifecycle hooks preserve caller-thread semantics outside hot paths.""" mgr = PluginManager() diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 9e82f368287d..fae1bb6da929 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -2218,10 +2218,10 @@ def test_timed_out_pre_tool_hook_blocks_before_tool_dispatch(self, agent, monkey manager = PluginManager() manager._hook_timeout_seconds = 0.02 + release = threading.Event() def slow_pre_tool_call(**kwargs): - import time - time.sleep(0.30) + release.wait(30) manager._hooks["pre_tool_call"] = [slow_pre_tool_call] monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager) @@ -2238,11 +2238,14 @@ def slow_pre_tool_call(**kwargs): side_effect=AssertionError("checkpoint should not run") ) - with patch("run_agent.handle_function_call", side_effect=AssertionError("should not run")): - agent._execute_tool_calls_sequential(mock_msg, messages, "task-1") + 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"] + 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( From fa038125b93c4cde91762aa39863c56aeca057e0 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 20 Aug 2026 04:46:54 -0400 Subject: [PATCH 06/10] fix(plugins): guard callback identity and timeout cleanup --- tests/hermes_cli/test_plugins.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 9e133e57fbac..924588702222 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -984,8 +984,10 @@ 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] @@ -996,6 +998,7 @@ def slow(**kwargs): 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() From 9a8ffe3360c4ea34d359a3f02a2ad66f6b927abe Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 20 Aug 2026 05:06:08 -0400 Subject: [PATCH 07/10] fix(plugins): serialize healthy callback admission Wait for a currently running callback to finish before admitting a concurrent invocation, while preserving timeout suppression for abandoned callbacks and clearing admission state during scoped unload. --- hermes_cli/plugins.py | 88 ++++++++++++++++++----- tests/agent/test_system_prompt_restore.py | 3 + tests/hermes_cli/test_plugins.py | 61 ++++++++++++++-- 3 files changed, 127 insertions(+), 25 deletions(-) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index a04bbf9c2584..1da367d4c395 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -3497,7 +3497,7 @@ def __init__(self, scope_key: Optional[str] = None) -> None: 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: Set[tuple] = set() + self._hook_running_callbacks: Dict[tuple, threading.Event] = {} self._hook_timeout_lock = threading.Lock() # ----------------------------------------------------------------------- @@ -3697,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) + if not unload_all and hook_names: + stale_events = [] + 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 @@ -5179,31 +5201,59 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: except Exception as exc: logger.warning("Hook '%s' callback %s raised: %s", hook_name, callback_name, exc) continue - now = time.monotonic() - with self._hook_timeout_lock: - suppressed_until = self._hook_timeout_suppressed_until.get(callback_key) - active = callback_key in self._hook_running_callbacks - if active or (suppressed_until is not None and suppressed_until > now): - logger.warning( - "Hook '%s' callback %s skipped while active or suppressed; continuing %s", - hook_name, - callback_name, - "fail-closed" if hook_name in _HOOK_TIMEOUT_FAIL_CLOSED_HOOKS else "fail-open", - ) - 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"}) + skip_reason = None + while True: + now = time.monotonic() + with self._hook_timeout_lock: + 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 - if suppressed_until is not None: - self._hook_timeout_suppressed_until.pop(callback_key, None) - self._hook_running_callbacks.add(callback_key) + 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 + 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: + 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) -> Any: + def bounded_callback( + callback=cb, + callback_key=callback_key, + context=context, + running_event=running_event, + ) -> Any: try: return context.run(self._invoke_hook_callback, callback, kwargs) finally: with self._hook_timeout_lock: - self._hook_running_callbacks.discard(callback_key) + if self._hook_running_callbacks.get(callback_key) is running_event: + self._hook_running_callbacks.pop(callback_key, None) + running_event.set() try: bounded_result = run_bounded_sync( bounded_callback, diff --git a/tests/agent/test_system_prompt_restore.py b/tests/agent/test_system_prompt_restore.py index cbbf8d5be091..c7c4f57a7511 100644 --- a/tests/agent/test_system_prompt_restore.py +++ b/tests/agent/test_system_prompt_restore.py @@ -17,6 +17,7 @@ import logging import threading +import time from unittest.mock import MagicMock import pytest @@ -158,6 +159,7 @@ def slow_session_start(**kwargs): db = MagicMock() agent = _make_agent(session_db=db) + started = time.perf_counter() try: _restore_or_build_system_prompt(agent, None, []) finally: @@ -165,6 +167,7 @@ def slow_session_start(**kwargs): 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 # --------------------------------------------------------------------------- diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 924588702222..f0692ea0f05b 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -783,6 +783,8 @@ 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, @@ -815,10 +817,6 @@ def test_hook_timeout_config_uses_positive_values_only(self, monkeypatch): assert plugins._get_hook_timeout_seconds() == plugins._DEFAULT_HOOK_TIMEOUT_SECONDS assert plugins._get_hook_timeout_seconds() == plugins._DEFAULT_HOOK_TIMEOUT_SECONDS - """Tests for lifecycle hook registration and invocation.""" - - - def test_pre_gateway_dispatch_collects_action_dicts(self, tmp_path, monkeypatch): """pre_gateway_dispatch callbacks return action dicts (skip/rewrite/allow).""" plugins_dir = tmp_path / "hermes_test" / "plugins" @@ -1002,6 +1000,42 @@ def slow(**kwargs): 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_bounded_hook_uses_shared_adapter_and_caller_context(self): mgr = PluginManager() marker = ContextVar("plugin-test-marker", default="missing") @@ -1082,12 +1116,27 @@ 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 - mgr._hook_running_callbacks.add(callback_key) + mgr._hook_running_callbacks[callback_key] = threading.Event() mgr.unload() assert mgr._hook_timeout_suppressed_until == {} - assert mgr._hook_running_callbacks == set() + assert mgr._hook_running_callbacks == {} + + 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_non_hot_hook_stays_on_caller_thread(self): """Lifecycle hooks preserve caller-thread semantics outside hot paths.""" From fc91b8462bc955d211ac4caeb5d084f66b085994 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 20 Aug 2026 05:22:29 -0400 Subject: [PATCH 08/10] fix(plugins): preserve callback ownership during teardown Keep healthy concurrent callbacks serialized, preserve shell-hook ownership of subprocess deadlines, and invalidate waiting callback snapshots when a plugin unloads. --- hermes_cli/plugins.py | 59 +++++++++++++++++------- tests/hermes_cli/test_plugins.py | 79 ++++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+), 16 deletions(-) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 1da367d4c395..1deff70edf56 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -5193,7 +5193,11 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: 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: + 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: @@ -5205,18 +5209,26 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: while True: now = time.monotonic() with self._hook_timeout_lock: - suppressed_until = self._hook_timeout_suppressed_until.get(callback_key) - if suppressed_until is not None and suppressed_until > now: - skip_reason = "suppressed" + 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: - 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 + 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): @@ -5226,8 +5238,9 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: self._hook_timeout_suppressed_until[callback_key] = ( time.monotonic() + self._hook_timeout_suppression_seconds ) - skip_reason = "active callback timed out" - break + 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", @@ -5236,7 +5249,10 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]: 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: + 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() @@ -5262,9 +5278,20 @@ def bounded_callback( ) if bounded_result.timed_out: with self._hook_timeout_lock: - self._hook_timeout_suppressed_until[callback_key] = ( - time.monotonic() + self._hook_timeout_suppression_seconds + 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 diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index f0692ea0f05b..c1eb24dd963d 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -1036,6 +1036,35 @@ def invoke(): 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") @@ -1138,6 +1167,56 @@ def test_scoped_unload_clears_removed_hook_timeout_state(self): 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() From 9fb70e4c99da9e9f031cbf87ada7c0ba934e6180 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 20 Aug 2026 05:41:49 -0400 Subject: [PATCH 09/10] fix(plugins): invalidate callbacks during unload Signal full-unload waiters and recheck callback ownership before a bounded worker enters plugin code, so removed callbacks cannot run or restore timeout state after teardown. --- hermes_cli/plugins.py | 17 ++++++++++++++++- tests/hermes_cli/test_plugins.py | 4 +++- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 1deff70edf56..f63492ad4584 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -3701,8 +3701,8 @@ def _unload_scoped( self._dispose_registrations(registrations) self._forget_registrations(registrations) + stale_events = [] if not unload_all and hook_names: - stale_events = [] with self._hook_timeout_lock: live_callback_keys = { (hook_name, id(callback)) @@ -3775,6 +3775,7 @@ def _unload_scoped( 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() @@ -3795,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 # ----------------------------------------------------------------------- @@ -5264,6 +5268,17 @@ def bounded_callback( 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: diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index c1eb24dd963d..eee538d45a71 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -1145,12 +1145,14 @@ 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 - mgr._hook_running_callbacks[callback_key] = threading.Event() + 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() From f8894c056392ab558fd4427d7fe620ade2458ee6 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 20 Aug 2026 05:56:13 -0400 Subject: [PATCH 10/10] ci: retrigger after runner timeout Retry the hosted workflow after the failed test slice timed out while resolving the CI uv manifest before tests started.