fix(plugins): bound background discovery handoff

hermes_cli/plugins.py

@@ -1563,8 +1563,11 @@ def has_enabled_agent_plugin_mcp(raw_config: Mapping[str, Any]) -> bool:
 
 def discover_plugins(force: bool = False) -> None:
     """Discover and load all plugins (idempotent; ``force=True`` rescans). Joins an in-flight
-    background discovery instead of racing a second scan."""
-    _join_background_discovery()
+    background discovery instead of racing a second scan. If that join times out, returns with
+    registrations loaded so far while the background scan continues."""
+    if not _join_background_discovery():
+        logger.warning("Plugin discovery is still running; continuing with partial plugin state")
+        return
     get_plugin_manager().discover_and_load(force=force)
 
 
@@ -1595,12 +1598,13 @@ def _run() -> None:
         _background_discovery_thread.start()
 
 
-def _join_background_discovery(timeout: float = 30.0) -> None:
-    """Wait for an in-flight background discovery (no-op from its own thread)."""
+def _join_background_discovery(timeout: float = 30.0) -> bool:
+    """Wait for in-flight background discovery; return whether it finished within ``timeout``."""
     t = _background_discovery_thread
     if t is None or not t.is_alive() or t is threading.current_thread():
-        return
+        return True
     t.join(timeout=timeout)
+    return not t.is_alive()
 
 
 def _plugin_toolset_keys_cache_path() -> Path:

tests/hermes_cli/test_plugins.py

@@ -122,6 +122,83 @@ def _make_plugin_dir(base: Path, name: str, *, register_body: str = "pass",
 class TestPluginDiscovery:
     """Tests for plugin discovery from directories and entry points."""
 
+    def test_discover_plugins_returns_when_background_discovery_exceeds_bound(
+        self, monkeypatch
+    ):
+        from hermes_cli import plugins as plugins_mod
+
+        manager = PluginManager()
+        discovery_started = threading.Event()
+        release_discovery = threading.Event()
+        discovery_finished = threading.Event()
+        caller_finished = threading.Event()
+        calls = 0
+
+        def _blocked_discovery(self_inner):
+            nonlocal calls
+            calls += 1
+            discovery_started.set()
+            release_discovery.wait(timeout=5)
+            discovery_finished.set()
+
+        monkeypatch.setattr(plugins_mod, "_plugin_manager", manager)
+        monkeypatch.setattr(plugins_mod, "_background_discovery_thread", None)
+        monkeypatch.setattr(PluginManager, "_discover_and_load_inner", _blocked_discovery)
+        original_join = plugins_mod._join_background_discovery
+        monkeypatch.setattr(
+            plugins_mod,
+            "_join_background_discovery",
+            lambda: original_join(timeout=0.05),
+        )
+
+        plugins_mod.start_background_plugin_discovery()
+        assert discovery_started.wait(timeout=2)
+        background_thread = plugins_mod._background_discovery_thread
+
+        caller = threading.Thread(
+            target=lambda: (plugins_mod.discover_plugins(), caller_finished.set()),
+            daemon=True,
+        )
+        caller.start()
+        try:
+            assert caller_finished.wait(timeout=2), (
+                "discover_plugins() re-entered the manager's locked discovery path"
+            )
+            assert background_thread is not None and background_thread.is_alive()
+            assert not discovery_finished.is_set()
+            assert calls == 1
+        finally:
+            release_discovery.set()
+            caller.join(timeout=2)
+            if background_thread is not None:
+                background_thread.join(timeout=2)
+
+        assert discovery_finished.is_set()
+        assert manager._discovered is True
+
+    def test_discover_plugins_preserves_healthy_background_completion(self, monkeypatch):
+        from hermes_cli import plugins as plugins_mod
+
+        manager = PluginManager()
+        discovery_finished = threading.Event()
+        calls = 0
+
+        def _healthy_discovery(self_inner):
+            nonlocal calls
+            calls += 1
+            discovery_finished.set()
+
+        monkeypatch.setattr(plugins_mod, "_plugin_manager", manager)
+        monkeypatch.setattr(plugins_mod, "_background_discovery_thread", None)
+        monkeypatch.setattr(PluginManager, "_discover_and_load_inner", _healthy_discovery)
+
+        plugins_mod.start_background_plugin_discovery()
+        plugins_mod.discover_plugins()
+
+        assert discovery_finished.is_set()
+        assert manager._discovered is True
+        assert calls == 1
+
     def test_removed_relay_plugin_identity_cannot_be_reloaded(
         self, monkeypatch, caplog
     ):