fix(plugins): don't drop healthy concurrent fail-open hook invocations

hermes_cli/config_defaults.py

@@ -2603,6 +2603,10 @@
         # callback. Shell hooks keep their own per-entry ``timeout``. Set to 0
         # to disable the cap (sync call on the agent thread). Capped at 600.
         "hook_callback_timeout": 30,
+        # Max concurrent in-flight workers per (hook, callback) for fail-open
+        # hooks before further invocations are dropped and counted as
+        # overflow rather than run unbounded. Capped at 64.
+        "hook_callback_max_concurrency": 4,
     },
 
     # Shell-script hooks — declarative bridge that invokes shell scripts

hermes_cli/plugins.py

@@ -3669,6 +3669,47 @@ def register_skill(
 _HOOK_CALLBACK_TIMEOUT_SECS = 30.0
 _MAX_HOOK_CALLBACK_TIMEOUT_SECS = 600.0
 
+# Cap on concurrent in-flight workers per (hook, callback) for fail-open
+# hooks, so a sustained event burst cannot fan out unbounded workers once a
+# callback's duration exceeds its arrival rate. Overridden by
+# ``plugins.hook_callback_max_concurrency``. Does not apply to fail-closed
+# hooks (``pre_tool_call``), which already skip on any overlap.
+_HOOK_CALLBACK_MAX_CONCURRENCY = 4
+_MAX_HOOK_CALLBACK_MAX_CONCURRENCY = 64
+
+
+def _resolve_hook_callback_max_concurrency() -> int:
+    """Return the effective per-callback concurrency cap for fail-open hooks.
+
+    Reads ``plugins.hook_callback_max_concurrency`` via the cached readonly
+    config loader. Falls back to ``_HOOK_CALLBACK_MAX_CONCURRENCY``. Values
+    ``< 1`` are clamped to 1; values above
+    ``_MAX_HOOK_CALLBACK_MAX_CONCURRENCY`` are clamped down.
+    """
+    limit = _HOOK_CALLBACK_MAX_CONCURRENCY
+    try:
+        from hermes_cli.config import load_config_readonly
+
+        plugins_cfg = (load_config_readonly() or {}).get("plugins")
+        if isinstance(plugins_cfg, dict) and "hook_callback_max_concurrency" in plugins_cfg:
+            raw = plugins_cfg.get("hook_callback_max_concurrency")
+            if raw is not None:
+                limit = int(raw)
+    except (TypeError, ValueError):
+        logger.warning(
+            "plugins.hook_callback_max_concurrency is not an int; using default %d",
+            _HOOK_CALLBACK_MAX_CONCURRENCY,
+        )
+        limit = _HOOK_CALLBACK_MAX_CONCURRENCY
+    except Exception:
+        limit = _HOOK_CALLBACK_MAX_CONCURRENCY
+
+    if limit < 1:
+        return 1
+    if limit > _MAX_HOOK_CALLBACK_MAX_CONCURRENCY:
+        return _MAX_HOOK_CALLBACK_MAX_CONCURRENCY
+    return limit
+
 
 def _resolve_hook_callback_timeout() -> float:
     """Return the effective hook-callback timeout in seconds.
@@ -3789,9 +3830,15 @@ def __init__(self, scope_key: Optional[str] = None) -> None:
         self._slack_action_handlers: List[tuple] = []
         # In-flight / recently-timed-out hook callbacks. Keyed by
         # (hook_name, id(cb)) so a stuck policy hook cannot spawn a new
-        # abandoned daemon thread on every subsequent fire.
-        self._hook_running_callbacks: Dict[tuple, object] = {}
+        # abandoned daemon thread on every subsequent fire. Each value is
+        # the set of tokens for workers currently in flight for that
+        # callback (fail-open hooks may have more than one concurrently).
+        self._hook_running_callbacks: Dict[tuple, Set[object]] = {}
         self._hook_timeout_suppressed_until: Dict[tuple, float] = {}
+        # Count of invocations dropped for exceeding
+        # ``hook_callback_max_concurrency`` — distinct from a confirmed
+        # timeout — keyed the same as ``_hook_running_callbacks``.
+        self._hook_overflow_counts: Dict[tuple, int] = {}
         self._hook_timeout_lock = threading.Lock()
         self._hook_timeout_suppression_seconds = _HOOK_TIMEOUT_SUPPRESSION_SECONDS
         # Registration handles are kept both per plugin (ownership lookup) and
@@ -4186,6 +4233,7 @@ def _unload_scoped(
             with self._hook_timeout_lock:
                 self._hook_running_callbacks.clear()
                 self._hook_timeout_suppressed_until.clear()
+                self._hook_overflow_counts.clear()
             self._discovered = False
         else:
             for key in target_keys:
@@ -5575,7 +5623,16 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]:
         ``plugins.hook_callback_timeout`` (default 30s). On timeout the worker
         is abandoned (not joined) so we do not reintroduce the #6622 hang.
         Timed-out or still-running ``pre_tool_call`` callbacks fail closed
-        with a block directive; other bounded hooks fail open (skip).
+        with a block directive. Other bounded hooks fail open: a confirmed
+        timeout is still skipped (bounding duplicate workers of a hung
+        callback), but a healthy invocation still in flight from an earlier
+        call runs concurrently on its own worker rather than being dropped —
+        up to ``plugins.hook_callback_max_concurrency`` (default 4) in-flight
+        workers per callback. Invocations beyond that cap are dropped and
+        counted in ``_hook_overflow_counts`` rather than mislabeled as a
+        timeout. Concurrent workers for the same callback may complete out
+        of order; callbacks that depend on ordering must serialize
+        themselves (e.g. via ``_HOOK_CALLER_THREAD_HOOKS``).
 
         ``subagent_stop`` (and any hook in ``_HOOK_CALLER_THREAD_HOOKS``)
         always runs on the caller thread to preserve the documented parent-
@@ -5614,26 +5671,62 @@ def invoke_hook(self, hook_name: str, **kwargs: Any) -> List[Any]:
                 if use_timeout:
                     token = object()
                     now = time.monotonic()
+                    max_concurrency = _resolve_hook_callback_max_concurrency()
                     with self._hook_timeout_lock:
                         suppressed_until = self._hook_timeout_suppressed_until.get(
                             callback_key
                         )
-                        running = callback_key in self._hook_running_callbacks
-                        if (
+                        timed_out = (
                             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,
-                            )
+                        )
+                        # A confirmed timeout must not spawn unbounded duplicates
+                        # of a hung worker, and pre_tool_call must fail closed on
+                        # any overlap. But "running" alone just means a healthy
+                        # invocation of a fail-open observer hook hasn't returned
+                        # yet — dropping it here silently loses telemetry (#98382).
+                        running_tokens = self._hook_running_callbacks.get(
+                            callback_key
+                        )
+                        running_count = len(running_tokens) if running_tokens else 0
+                        running = running_count > 0
+                        # Fail-open hooks still bound total concurrent workers
+                        # per callback so a sustained burst can't fan out
+                        # unboundedly once callback duration exceeds arrival
+                        # rate. This is bookkept separately from a confirmed
+                        # timeout so queue saturation is never mislabeled as
+                        # a hang.
+                        overflow = (
+                            not fail_closed and running_count >= max_concurrency
+                        )
+                        if timed_out or (running and fail_closed) or overflow:
+                            if overflow and not timed_out:
+                                self._hook_overflow_counts[callback_key] = (
+                                    self._hook_overflow_counts.get(callback_key, 0)
+                                    + 1
+                                )
+                                logger.warning(
+                                    "Hook '%s' callback %s skipped: %d worker(s) "
+                                    "already in flight (max concurrency %d)",
+                                    hook_name,
+                                    callback_name,
+                                    running_count,
+                                    max_concurrency,
+                                )
+                            else:
+                                logger.warning(
+                                    "Hook '%s' callback %s skipped after previous "
+                                    "timeout or while still running",
+                                    hook_name,
+                                    callback_name,
+                                )
                             if fail_closed:
                                 results.append(_pre_tool_call_timeout_block())
                             continue
                         if suppressed_until is not None:
                             self._hook_timeout_suppressed_until.pop(callback_key, None)
-                        self._hook_running_callbacks[callback_key] = token
+                        self._hook_running_callbacks.setdefault(
+                            callback_key, set()
+                        ).add(token)
 
                     context = contextvars.copy_context()
                     done = threading.Event()
@@ -5656,8 +5749,11 @@ def _runner(
                             failure["exc"] = exc
                         finally:
                             with self._hook_timeout_lock:
-                                if self._hook_running_callbacks.get(_key) is _token:
-                                    self._hook_running_callbacks.pop(_key, None)
+                                tokens = self._hook_running_callbacks.get(_key)
+                                if tokens is not None:
+                                    tokens.discard(_token)
+                                    if not tokens:
+                                        self._hook_running_callbacks.pop(_key, None)
                             done.set()
 
                     thread = threading.Thread(

tests/hermes_cli/test_plugins.py

@@ -1135,6 +1135,103 @@ def blocker(**_kwargs):
         assert elapsed < 1.0
         hold.set()
 
+    def test_concurrent_healthy_invocations_both_processed(self, monkeypatch):
+        """Two healthy overlapping calls to a fail-open hook must both run.
+
+        Regression for #98382: ``running`` was treated the same as a
+        confirmed timeout, so a second concurrent invocation of an observer
+        hook was silently dropped even though the first callback was well
+        within its timeout budget.
+        """
+        import time
+
+        monkeypatch.setattr(
+            "hermes_cli.plugins._resolve_hook_callback_timeout", lambda: 5.0
+        )
+
+        calls = []
+        lock = threading.Lock()
+
+        def slow(**kwargs):
+            time.sleep(0.3)
+            with lock:
+                calls.append(kwargs.get("marker"))
+            return kwargs.get("marker")
+
+        mgr = PluginManager()
+        mgr._hooks["post_tool_call"] = [slow]
+
+        results = {}
+
+        def _invoke(marker):
+            results[marker] = mgr.invoke_hook("post_tool_call", marker=marker)
+
+        t1 = threading.Thread(target=_invoke, args=("first",))
+        t1.start()
+        time.sleep(0.15)
+        t2 = threading.Thread(target=_invoke, args=("second",))
+        t2.start()
+        t1.join(timeout=5.0)
+        t2.join(timeout=5.0)
+
+        assert sorted(calls) == ["first", "second"]
+        assert results["first"] == ["first"]
+        assert results["second"] == ["second"]
+
+    def test_burst_beyond_max_concurrency_is_bounded_and_counted(self, monkeypatch):
+        """A burst past ``hook_callback_max_concurrency`` is bounded, not fanned out.
+
+        Follow-up to #98382/#98385: fail-open concurrency is now capped per
+        callback so a sustained burst faster than callback duration can't
+        spawn unbounded workers. Overflow must be counted separately from a
+        confirmed timeout (``_hook_timeout_suppressed_until`` stays empty).
+        """
+        import time
+
+        monkeypatch.setattr(
+            "hermes_cli.plugins._resolve_hook_callback_timeout", lambda: 5.0
+        )
+        monkeypatch.setattr(
+            "hermes_cli.plugins._resolve_hook_callback_max_concurrency", lambda: 2
+        )
+
+        in_flight = []
+        max_in_flight = []
+        lock = threading.Lock()
+
+        def slow(**_kwargs):
+            with lock:
+                in_flight.append(1)
+                max_in_flight.append(len(in_flight))
+            time.sleep(0.3)
+            with lock:
+                in_flight.pop()
+            return "ran"
+
+        mgr = PluginManager()
+        mgr._hooks["post_tool_call"] = [slow]
+        callback_key = ("post_tool_call", id(slow))
+
+        results = {}
+
+        def _invoke(marker):
+            results[marker] = mgr.invoke_hook("post_tool_call")
+
+        threads = [
+            threading.Thread(target=_invoke, args=(i,)) for i in range(4)
+        ]
+        for t in threads:
+            t.start()
+            time.sleep(0.05)
+        for t in threads:
+            t.join(timeout=5.0)
+
+        assert max(max_in_flight) <= 2
+        assert sum(r == ["ran"] for r in results.values()) == 2
+        assert sum(r == [] for r in results.values()) == 2
+        assert mgr._hook_overflow_counts.get(callback_key) == 2
+        assert callback_key not in mgr._hook_timeout_suppressed_until
+
     def test_pre_tool_call_timeout_fail_closed(self, monkeypatch):
         """Timed-out pre_tool_call must return a block directive, not allow."""
         import time