fix(plugins): per-plugin load deadline so a hung register() no longer hangs startup

A plugin whose import or register() never returns (an infinite loop, a blocking
network call) held PluginManager.discover_and_load() forever, and with it every
synchronous caller: `hermes chat`, gateway startup, ACP session/new (#108139).

Each plugin's import + register() now runs under `plugins.load_timeout_seconds`
(default 10, 0 disables, max 600) on a daemon worker. On overrun the plugin is
recorded as failed with "load timed out after Ns" (same channel as every other
load failure: startup WARNING, `/plugins`, `list_plugins()`), its pre-hang
registrations are disposed, and discovery continues with the next plugin. The
abandoned worker's later `ctx.register_*`/`subscribe`/`on_unload` calls are
refused with a WARNING (the context is marked abandoned), so a late registration
can never land in a registry the failure path already swept. Abandoned loaders
are capped per process (8); past the cap further loads are refused with a named
reason rather than run inline, which would recreate the hang (#98382 shape).

Because the worker cannot own the caller's RLocks: the deferred-platform eager
fallback now runs outside the replacement transaction, discovery re-entered from
a loader worker returns on the already-set discovered flag instead of blocking on
the sweep's lock, and such a worker never joins the background discovery thread
that is waiting on it.

hermes_cli/config_defaults.py

@@ -1683,6 +1683,10 @@ DEFAULT_CONFIG = {
         # Wall-clock cap (seconds) for one in-process Python plugin hook callback; shell hooks keep
         # their own per-entry `timeout`. 0 = no cap (sync call on agent thread). Max 600.
         "hook_callback_timeout": 30,
+        # Deadline (seconds) for one plugin's import + register() at load. A plugin that overruns it is
+        # skipped with the reason "load timed out" and the rest keep loading; the stuck worker thread is
+        # abandoned. 0 = no deadline (load inline). Max 600.
+        "load_timeout_seconds": 10,
         # Keep loading external plugins that still import pre-decomposition module paths after the
         # 2026-09-14 removal date (see COMPAT_MANIFEST.md, `hermes plugins compat`). Stopgap only: the
         # old paths raise ImportError once the compat layer is actually removed.

hermes_cli/plugins.py

@@ -24,7 +24,7 @@ import threading
 import types
 from contextlib import suppress
 from dataclasses import dataclass, field
-from functools import cached_property
+from functools import cached_property, wraps
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Mapping, Optional, Set, Tuple, Union
 
@@ -47,7 +47,7 @@ from hermes_cli.plugins_discovery import (  # noqa: F401 — re-exported
 )
 from hermes_cli.plugins_loader import (
     PluginLoaderMixin, _BARE_MODULE_SCOPE, _MODULE_NAMESPACE_LOCK, _NS_PARENT, _evict_modules,
-    _plugin_home_scope, _serialized_replacement,
+    _plugin_home_scope, _serialized_replacement, in_plugin_load_worker,
 )
 from hermes_cli.plugins_dispatch import (  # noqa: F401 — re-exported
     DEFAULT_SYSTEM_PROMPT_SECTION_MAX_CHARS, HERMES_EVENT_NAMESPACE, MAX_SYSTEM_PROMPT_SECTION_CHARS,
@@ -229,6 +229,13 @@ class PluginContext:
         self.manifest = manifest
         self._manager = manager
         self._llm: Any = None  # lazy; tests preseed it (see ``llm``)
+        # Set when this context's load overran ``plugins.load_timeout_seconds``: the abandoned worker may
+        # still be running register(), and nothing it registers from then on may reach a registry.
+        self._load_abandoned = False
+
+    def _abandon_load(self) -> None:
+        """Mark this load as timed out; every later ``register_*``/``subscribe``/``on_unload`` is ignored."""
+        self._load_abandoned = True
 
     @property
     def plugin_id(self) -> str:
@@ -1110,6 +1117,31 @@ for _row in _SCOPED_PROVIDER_REGISTRARS:
 del _row
 
 
+def _ignore_after_abandoned_load(method):
+    """Turn a registrar into a no-op once the context's load timed out: the abandoned worker thread may
+    still be executing register(), and a late registration would land in registries that the failure
+    path already swept (#108139)."""
+    @wraps(method)
+    def wrapped(self, *args, **kwargs):
+        if getattr(self, "_load_abandoned", False):
+            logger.warning(
+                "Plugin '%s' called %s() after its load timed out; ignored", self.manifest.name,
+                method.__name__,
+            )
+            return None
+        return method(self, *args, **kwargs)
+
+    return wrapped
+
+
+# Every mutating entry point plugins reach through ``ctx`` during register(); applied by name so the
+# guard cannot drift from the surface as registrars are added.
+for _name, _method in list(vars(PluginContext).items()):
+    if callable(_method) and (_name.startswith("register_") or _name in {"subscribe", "on_unload"}):
+        setattr(PluginContext, _name, _ignore_after_abandoned_load(_method))
+del _name, _method
+
+
 def _resolve_hook_callback_timeout() -> float:
     """Effective hook-callback timeout from ``plugins.hook_callback_timeout`` (default 30s; ``<= 0``
     disables the threaded path; clamped to ``_MAX_HOOK_CALLBACK_TIMEOUT_SECS``)."""
@@ -1232,6 +1264,11 @@ class PluginManager(PluginLoaderMixin, PluginDispatchMixin, PluginLedgerMixin):
     def discover_and_load(self, force: bool = False) -> None:
         """Scan all plugin sources and load each plugin found; ``force`` unloads first so config
         changes / new bundled backends become visible in long-lived sessions."""
+        if self._discovered and not force and in_plugin_load_worker():
+            # A plugin whose register() re-enters discovery (importing model_tools does) runs on a
+            # deadline worker that cannot re-acquire the sweep's RLock; the flag is already set for the
+            # whole sweep, so return where the locked re-entry used to. Every other caller still waits.
+            return
         with self._discovery_lock, _plugin_home_scope(self.home_path):
             if self._discovered and not force:
                 return
@@ -1631,9 +1668,10 @@ def start_background_plugin_discovery() -> None:
 
 
 def _join_background_discovery(timeout: float = 30.0) -> None:
-    """Wait for an in-flight background discovery (no-op from its own thread)."""
+    """Wait for an in-flight background discovery (no-op from its own thread or a plugin-load worker it
+    spawned — that worker's parent is blocked waiting on it)."""
     t = _background_discovery_thread
-    if t is None or not t.is_alive() or t is threading.current_thread():
+    if t is None or not t.is_alive() or t is threading.current_thread() or in_plugin_load_worker():
         return
     t.join(timeout=timeout)
 

hermes_cli/plugins_loader.py

@@ -7,6 +7,7 @@ through ``hermes_cli.plugins`` so tests that patch them on the origin keep worki
 
 from __future__ import annotations
 
+import contextvars
 import hashlib
 import importlib
 import importlib.metadata
@@ -28,7 +29,7 @@ from hermes_cli.plugins_manifest import PluginManifest, manifest_key, validate_c
 from hermes_cli.plugins_state import _plugin_settings_entry
 
 if TYPE_CHECKING:  # pragma: no cover
-    from hermes_cli.plugins import LoadedPlugin
+    from hermes_cli.plugins import LoadedPlugin, PluginContext
 
 logger = logging.getLogger("hermes_cli.plugins")
 
@@ -36,6 +37,102 @@ _NS_PARENT = "hermes_plugins"
 _MODULE_NAMESPACE_LOCK = threading.RLock()
 _BARE_MODULE_SCOPE: Dict[str, str] = {}  # bare module name -> owning scope_key
 
+# Per-plugin deadline on import + register(): ``plugins.load_timeout_seconds`` (default 10s, 0 disables,
+# clamped to the max). A plugin that never returns is skipped with a named reason and loading moves on
+# (#108139). Python cannot kill a thread, so the worker is abandoned as a daemon; the cap bounds how many
+# abandoned loaders one process may accumulate (#98382) — past it, further loads are refused, not run inline.
+_LOAD_TIMEOUT_SECS = 10.0
+_MAX_LOAD_TIMEOUT_SECS = 600.0
+_MAX_ABANDONED_LOADERS = 8
+_ABANDONED_LOADERS: List[threading.Thread] = []
+_ABANDONED_LOADERS_LOCK = threading.Lock()
+_IN_PLUGIN_LOAD = threading.local()  # ``.active`` on a loader worker thread
+
+
+class PluginLoadTimeout(Exception):
+    """Raised on the loading thread when a plugin's import + ``register()`` overran its deadline."""
+
+
+def in_plugin_load_worker() -> bool:
+    """True on a deadline worker thread; re-entrant discovery from there must not block on its own parent."""
+    return bool(getattr(_IN_PLUGIN_LOAD, "active", False))
+
+
+def _resolve_plugin_load_timeout() -> float:
+    """Effective per-plugin load deadline from ``plugins.load_timeout_seconds`` (default 10s; ``0`` runs
+    loads inline with no deadline; clamped to ``_MAX_LOAD_TIMEOUT_SECS``)."""
+    default = _LOAD_TIMEOUT_SECS
+    try:
+        from hermes_cli.config import load_config_readonly
+        plugins_cfg = (load_config_readonly() or {}).get("plugins")
+        if not isinstance(plugins_cfg, dict) or plugins_cfg.get("load_timeout_seconds") is None:
+            return default
+        timeout = float(plugins_cfg["load_timeout_seconds"])
+    except (TypeError, ValueError):
+        logger.warning("plugins.load_timeout_seconds is not a number; using default %gs", default)
+        return default
+    except Exception:
+        return default
+    if timeout < 0:
+        logger.warning("plugins.load_timeout_seconds=%g is negative; using default %gs", timeout, default)
+        return default
+    if timeout > _MAX_LOAD_TIMEOUT_SECS:
+        logger.warning("plugins.load_timeout_seconds=%g exceeds max %gs; clamping", timeout,
+                       _MAX_LOAD_TIMEOUT_SECS)
+        return _MAX_LOAD_TIMEOUT_SECS
+    return timeout
+
+
+def _reserve_abandoned_loader_slot() -> None:
+    """Drop finished abandoned loaders; refuse the load once the live cap is reached. Refusing beats
+    loading inline: at the cap the process already holds several hung loaders, so an inline load is the
+    exact startup hang this deadline exists to prevent."""
+    with _ABANDONED_LOADERS_LOCK:
+        _ABANDONED_LOADERS[:] = [t for t in _ABANDONED_LOADERS if t.is_alive()]
+        if len(_ABANDONED_LOADERS) < _MAX_ABANDONED_LOADERS:
+            return
+    raise PluginLoadTimeout(
+        f"not loaded: {_MAX_ABANDONED_LOADERS} abandoned plugin loader thread(s) are still running "
+        f"(plugins.load_timeout_seconds); restart Hermes to retry"
+    )
+
+
+def run_with_load_deadline(plugin_key: str, ctx: "PluginContext", fn: Callable[[], Any]) -> Any:
+    """Run ``fn`` (a plugin's import + ``register()``) under the per-plugin deadline.
+
+    The worker inherits the caller's context (the Hermes-home override is a ContextVar). On timeout the
+    worker is abandoned as a daemon, ``ctx`` is marked so any registration it still attempts is ignored,
+    and :class:`PluginLoadTimeout` is raised on the calling thread so the usual failure path records the
+    reason and disposes whatever was registered before the hang.
+    """
+    timeout = _resolve_plugin_load_timeout()
+    if timeout <= 0:
+        return fn()
+    _reserve_abandoned_loader_slot()
+    outcome: List[Any] = []
+    failure: List[BaseException] = []
+
+    def _worker() -> None:
+        _IN_PLUGIN_LOAD.active = True
+        try:
+            outcome.append(fn())
+        except BaseException as exc:  # re-raised on the loading thread, KeyboardInterrupt included
+            failure.append(exc)
+
+    worker = threading.Thread(
+        target=contextvars.copy_context().run, args=(_worker,), name=f"plugin-load:{plugin_key}", daemon=True,
+    )
+    worker.start()
+    worker.join(timeout)
+    if worker.is_alive():
+        ctx._abandon_load()
+        with _ABANDONED_LOADERS_LOCK:
+            _ABANDONED_LOADERS.append(worker)
+        raise PluginLoadTimeout(f"load timed out after {timeout:g}s (import + register() never returned)")
+    if failure:
+        raise failure[0]
+    return outcome[0]
+
 
 def _evict_modules(module_name: str) -> None:
     """Drop ``module_name`` and every ``module_name.*`` submodule from ``sys.modules``."""
@@ -95,16 +192,26 @@ class PluginLoaderMixin:
             return name[: -len("-platform")]
         return Path(manifest.path).name if manifest.path else name
 
-    @_serialized_replacement
     def _register_deferred_platform(self, manifest: PluginManifest) -> None:
         """Register a lazy loader for a bundled platform: the adapter imports only when the
         ``platform_registry`` is first asked for it; a placeholder ``LoadedPlugin`` keeps it visible in
         ``hermes plugins list`` until then."""
         from hermes_cli.plugins import LoadedPlugin
         lookup_key = manifest_key(manifest)
-        platform_name = self._platform_name_from_manifest(manifest)
         loaded = LoadedPlugin(manifest=manifest, enabled=True, deferred=True)
         self._plugins[lookup_key] = loaded
+        if not self._lease_deferred_platform(manifest, lookup_key):
+            # Fall back to eager loading so the platform is never silently lost. Runs outside the
+            # replacement transaction: the eager load's register() executes on a deadline worker, whose
+            # registrations need the coordinator lock this thread would otherwise still hold.
+            self._load_plugin(manifest)
+            return
+        self._register_deferred_platform_tools(manifest, loaded)
+
+    @_serialized_replacement
+    def _lease_deferred_platform(self, manifest: PluginManifest, lookup_key: str) -> bool:
+        """Publish the deferred loader as a ledger-owned lease; False when the registry refused it."""
+        platform_name = self._platform_name_from_manifest(manifest)
         try:
             from gateway.platform_registry import platform_registry
             scope = self.scope_key
@@ -128,12 +235,10 @@ class PluginLoaderMixin:
                 )
             logger.debug("Registered deferred platform loader: %s (plugin=%s)", platform_name, lookup_key)
         except Exception:
-            # Fall back to eager loading so the platform is never silently lost.
             logger.debug(
                 "Deferred platform registration failed for '%s'; eager-loading", lookup_key, exc_info=True)
-            self._load_plugin(manifest)
-            return
-        self._register_deferred_platform_tools(manifest, loaded)
+            return False
+        return True
 
     def _register_deferred_platform_tools(self, manifest: PluginManifest, loaded: LoadedPlugin) -> None:
         """Register a deferred platform's *client* tools without its adapter. Deferring the plugin would
@@ -301,7 +406,10 @@ class PluginLoaderMixin:
         registration_start = len(self._registration_order)
         module_name = self._policy_module_name(manifest)
         self._track_tool_override_policy(manifest, module_name)
-        try:
+        ctx = PluginContext(manifest, self)
+
+        def _import_and_register() -> bool:
+            """Import + register() — the part a plugin controls, so the part the deadline covers."""
             # Reuse a deferred platform's already-imported package so its body doesn't run twice.
             # See #78050.
             module = self._predeclared_modules.pop(plugin_key, None)
@@ -321,16 +429,22 @@ class PluginLoaderMixin:
             if register_fn is None:
                 loaded.error = "no register() function"
                 logger.warning("Plugin '%s' has no register() function", manifest.name)
-            else:
-                register_fn(PluginContext(manifest, self))
+                return False
+            register_fn(ctx)
+            return True
+
+        try:
+            if run_with_load_deadline(plugin_key, ctx, _import_and_register):
                 self._attribute_registrations(loaded, plugin_key, registration_start)
                 loaded.enabled = True
                 from hermes_cli.plugins_ledger import _hook_source_of
 
-                self._drop_fallback_hooks(_hook_source_of(manifest.name, module))
+                self._drop_fallback_hooks(_hook_source_of(manifest.name, loaded.module))
         except (Exception, SystemExit) as exc:
             # SystemExit too: a plugin module with an unguarded ``main()``/``sys.exit()`` must not take the
             # whole process (and every other plugin's registry) down with it; KeyboardInterrupt still propagates.
+            # PluginLoadTimeout lands here as well: the abandoned worker's later registrations are refused
+            # by ``ctx``, and whatever it registered before hanging is disposed below.
             owned = [r for r in self._registration_order if r.plugin_key == plugin_key]
             self._dispose_registrations(owned)
             self._forget_registrations(owned)

hermes_cli/web_server_config.py

@@ -170,6 +170,14 @@ _SCHEMA_OVERRIDES: Dict[str, Dict[str, Any]] = {
             "subagent_stop are never moved onto a timeout worker."
         ),
     },
+    "plugins.load_timeout_seconds": {
+        "type": "number",
+        "description": (
+            "Deadline (seconds) for one plugin's import + register() at load. A plugin that "
+            "overruns it is skipped with the reason 'load timed out' and the rest keep loading. "
+            "0 disables the deadline; values above 600 are clamped."
+        ),
+    },
 }
 
 # Small categories fold into a bigger tab to avoid one-field orphan tabs. Several sources

tests/hermes_cli/test_plugin_manifest_v2.py

@@ -479,6 +479,53 @@ class TestLoadIsolation:
         with pytest.raises(KeyboardInterrupt):
             PluginManager().discover_and_load()
 
+    def test_register_overrunning_load_timeout_skips_only_that_plugin(self, hermes_home, caplog):
+        """A register() that never returns used to hang startup forever (#108139). Under
+        ``plugins.load_timeout_seconds`` that plugin alone is recorded as failed with a named reason, its
+        pre-hang registrations are disposed, later plugins still load, and anything the abandoned worker
+        registers afterwards is ignored."""
+        import sys
+        import threading
+        sys._deadline_gate, sys._deadline_done = threading.Event(), threading.Event()
+        _write_plugin(hermes_home / "plugins", "b_slow", register_body=(
+            "import sys; ctx.register_hook('pre_tool_call', lambda **kw: None); sys._deadline_gate.wait(5); "
+            "ctx.register_hook('post_tool_call', lambda **kw: None); sys._deadline_done.set()"))
+        _write_plugin(hermes_home / "plugins", "c_after")
+        _enable(hermes_home, ["b_slow", "c_after"])
+        (hermes_home / "config.yaml").write_text(yaml.safe_dump(
+            {"plugins": {"enabled": ["b_slow", "c_after"], "load_timeout_seconds": 0.3}}))
+        mgr = PluginManager()
+        try:
+            with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"):
+                mgr.discover_and_load()
+                assert mgr._plugins["c_after"].enabled
+                assert not mgr._plugins["b_slow"].enabled
+                assert "load timed out after 0.3s" in (mgr._plugins["b_slow"].error or "")
+                assert mgr._hooks.get("pre_tool_call", []) == []  # registered before the hang → disposed
+                sys._deadline_gate.set()  # release the abandoned worker; its late registration must bounce
+                assert sys._deadline_done.wait(5)
+            assert mgr._hooks.get("post_tool_call", []) == []
+            assert "called register_hook() after its load timed out; ignored" in caplog.text
+        finally:
+            del sys._deadline_gate, sys._deadline_done
+
+    def test_load_timeout_zero_runs_register_inline(self, hermes_home):
+        """``plugins.load_timeout_seconds: 0`` disables the deadline: register() runs on the calling thread."""
+        import sys
+        import threading
+        _write_plugin(hermes_home / "plugins", "inline",
+                      register_body="import sys, threading; sys._load_thread = threading.current_thread()")
+        (hermes_home / "config.yaml").write_text(yaml.safe_dump(
+            {"plugins": {"enabled": ["inline"], "load_timeout_seconds": 0}}))
+        try:
+            mgr = PluginManager()
+            mgr.discover_and_load()
+            assert mgr._plugins["inline"].enabled
+            assert sys._load_thread is threading.current_thread()
+        finally:
+            if hasattr(sys, "_load_thread"):
+                del sys._load_thread
+
 
 class TestBundledKeyShadowing:
     def test_impostor_dir_cannot_claim_a_bundled_key(self, tmp_path, monkeypatch, caplog):

website/docs/user-guide/features/plugins.md

@@ -168,6 +168,13 @@ plugins:
   # subagent_stop are never moved onto a timeout worker.
   # Shell hooks keep their own per-entry timeout under the top-level hooks: key.
   hook_callback_timeout: 30
+  # Optional: deadline (seconds) for one plugin's import + register() at load.
+  # A plugin that overruns it is skipped with the reason "load timed out after
+  # Ns" (reported like any other load failure: the startup warning and the
+  # in-session `/plugins` listing) and the remaining plugins keep loading; the
+  # stuck thread is abandoned. Default 10; set 0 to disable; values above 600
+  # are clamped.
+  load_timeout_seconds: 10
 ```
 
 Three ways to flip state: