From 0c8cf21882bead813b0b2104e0eb04bb2d415c10 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 6 Jul 2026 14:59:36 -0600 Subject: [PATCH] feat(nemo-relay): activate dynamic plugins Signed-off-by: Bryan Bednarski --- plugins/observability/nemo_relay/README.md | 48 ++++- plugins/observability/nemo_relay/__init__.py | 171 +++++++++++++++-- pyproject.toml | 2 +- tests/plugins/test_nemo_relay_plugin.py | 189 ++++++++++++++++++- 4 files changed, 388 insertions(+), 22 deletions(-) diff --git a/plugins/observability/nemo_relay/README.md b/plugins/observability/nemo_relay/README.md index ddc27e491103..e3bd2dd441bc 100644 --- a/plugins/observability/nemo_relay/README.md +++ b/plugins/observability/nemo_relay/README.md @@ -84,14 +84,15 @@ wheel from this checkout, then install the official NeMo Relay runtime extra: ```bash uv build --wheel python -m pip install --force-reinstall dist/hermes_agent-*.whl -python -m pip install "nemo-relay==0.3" +python -m pip install "nemo-relay>=0.5,<0.7" hermes plugins enable observability/nemo_relay ``` -The plugin fails open when `nemo-relay` is not installed. Install and test it against the official NeMo Relay 0.3 PyPI distribution: +The plugin fails open when `nemo-relay` is not installed. Install a supported +NeMo Relay 0.5 or 0.6 distribution: ```bash -pip install "nemo-relay==0.3" +pip install "nemo-relay>=0.5,<0.7" ``` ## Export Configuration @@ -189,16 +190,46 @@ session, turn, approval, and subagent marks; the plugin skips its manual NeMo Relay. `tool_parallelism.mode = "observe_only"` keeps tool scheduling observational while still wrapping the real execution boundary. +### Dynamic Plugins (NeMo Relay 0.6) + +Hermes can activate explicit native or worker plugins through the NeMo Relay +0.6 binding. Add `dynamic_plugins` entries to the same file; each entry uses +the binding's `plugin_id`, `kind`, `manifest_ref`, optional `environment_ref`, +and component `config` fields: + +```toml +[[dynamic_plugins]] +plugin_id = "example-plugin" +kind = "rust_dynamic" +manifest_ref = "/absolute/path/to/example-plugin/relay-plugin.toml" + +[dynamic_plugins.config] +mode = "enabled" +``` + +Hermes activates these plugins before registering its managed LLM and tool +execution middleware and retains the activation for the runtime lifetime. +During shutdown it closes session exporters, flushes Relay subscribers, and +then closes the activation so callbacks are removed before plugin code is +unloaded. + +NeMo Relay 0.5 does not expose dynamic activation through its Python binding. +When `dynamic_plugins` is present with a 0.5 runtime, Hermes logs an actionable +warning and continues with the ordinary static component configuration, so +ATOF and ATIF observability remain available. No dynamic plugin is loaded in +that degraded mode. + For the full generic Hermes middleware contract, see [`docs/middleware/README.md`](../../../docs/middleware/README.md). ## Canonical Local Examples -The observe-only examples in this section use the official `nemo-relay==0.3` -distribution and a local Ollama model served through the OpenAI-compatible API. +The observe-only examples in this section use a supported NeMo Relay 0.5 or +0.6 distribution and a local Ollama model served through the OpenAI-compatible +API. ```bash -pip install "nemo-relay==0.3" +pip install "nemo-relay>=0.5,<0.7" export HERMES_HOME=/tmp/hermes-nemo-relay-docs/hermes-home mkdir -p "$HERMES_HOME" @@ -444,9 +475,8 @@ for the same execution. This example enables both NeMo Relay observability export and adaptive execution middleware for a local Hermes run. This path requires a NeMo Relay runtime that -supports `[components.config.tool_parallelism]`; the `nemo-relay==0.3` -install used by the earlier observability-only examples does not support this -adaptive config. +supports `[components.config.tool_parallelism]`, as provided by the supported +0.5 and 0.6 releases. ```bash export HERMES_HOME=/tmp/hermes-middleware-test/hermes-home diff --git a/plugins/observability/nemo_relay/__init__.py b/plugins/observability/nemo_relay/__init__.py index 4716b73ec040..774d2712c5f6 100644 --- a/plugins/observability/nemo_relay/__init__.py +++ b/plugins/observability/nemo_relay/__init__.py @@ -2,6 +2,7 @@ from __future__ import annotations +import atexit import asyncio import inspect import json @@ -44,6 +45,7 @@ class _SubagentParent: class _Settings: plugins_toml_path: str = "" plugins_config: dict[str, Any] | None = None + dynamic_plugins: list[dict[str, Any]] = field(default_factory=list) adaptive_enabled: bool = False adaptive_mode: str = "observe_only" atof_enabled: bool = False @@ -67,6 +69,8 @@ class _Runtime: self.subagent_parents: dict[str, _SubagentParent] = {} self.atof_exporter: Any = None self._atof_subscriber_name = "hermes.nemo_relay.atof" + self._plugin_activation: Any = None + self._shutdown_registered = False self._plugin_config_initialized = self._configure_plugins_toml() self._plugin_config_needs_reinit = False if not self._plugin_config_initialized: @@ -76,27 +80,64 @@ class _Runtime: if not self.settings.plugins_config: return False plugin_mod = getattr(self.nemo_relay, "plugin", None) + if plugin_mod is None: + return False + plugin_config = _static_plugin_config(self.settings.plugins_config) + if self.settings.dynamic_plugins: + activate_dynamic = getattr(plugin_mod, "activate_dynamic_plugins", None) + if callable(activate_dynamic): + try: + self._ensure_plugin_config_output_dirs(plugin_config) + self._plugin_activation = _resolve_awaitable( + activate_dynamic(plugin_config, self.settings.dynamic_plugins) + ) + self._ensure_shutdown_registered() + return True + except Exception as exc: + logger.warning( + "NeMo Relay dynamic plugin activation failed; continuing with static " + "observability only: %s", + exc, + ) + else: + logger.warning( + "NeMo Relay dynamic plugins require nemo-relay>=0.6,<0.7; this binding does " + "not expose plugin.activate_dynamic_plugins. Continuing with static " + "observability only." + ) initialize = getattr(plugin_mod, "initialize", None) if not callable(initialize): return False try: - self._ensure_plugin_config_output_dirs(self.settings.plugins_config) - _resolve_awaitable(initialize(self.settings.plugins_config)) + self._ensure_plugin_config_output_dirs(plugin_config) + _resolve_awaitable(initialize(plugin_config)) return True except Exception as exc: logger.debug("NeMo Relay plugins.toml init failed: %s", exc, exc_info=True) return False + def _ensure_shutdown_registered(self) -> None: + if self._shutdown_registered: + return + atexit.register(self.shutdown) + self._shutdown_registered = True + def _clear_plugins_toml(self) -> None: if not self._plugin_config_initialized: return - plugin_mod = getattr(self.nemo_relay, "plugin", None) - clear = getattr(plugin_mod, "clear", None) - if not callable(clear): - return try: - _resolve_awaitable(clear()) + if self._plugin_activation is not None: + _flush_relay_subscribers(self.nemo_relay) + close = getattr(self._plugin_activation, "close", None) + if callable(close): + _resolve_awaitable(close()) + else: + plugin_mod = getattr(self.nemo_relay, "plugin", None) + clear = getattr(plugin_mod, "clear", None) + if callable(clear): + _resolve_awaitable(clear()) finally: + self._plugin_activation = None self._plugin_config_initialized = False self._plugin_config_needs_reinit = bool(self.settings.plugins_config) @@ -226,20 +267,46 @@ class _Runtime: self.nemo_relay.scope.pop(state.handle, output=_jsonable(kwargs)) except Exception: logger.debug("NeMo Relay session pop failed", exc_info=True) + try: + _flush_relay_subscribers(self.nemo_relay) + except Exception: + logger.debug("NeMo Relay subscriber flush failed", exc_info=True) self.export_atif(state) if state.atif_exporter is not None and state.atif_subscriber_name: try: state.atif_exporter.deregister(state.atif_subscriber_name) except Exception: logger.debug("NeMo Relay ATIF deregister failed", exc_info=True) - if self._plugin_config_initialized and not self.sessions: + if ( + self._plugin_config_initialized + and self._plugin_activation is None + and not self.sessions + ): try: self._clear_plugins_toml() except Exception: logger.debug("NeMo Relay plugins.toml clear failed", exc_info=True) - elif self.settings.plugins_config and not self.sessions: + elif ( + self.settings.plugins_config + and self._plugin_activation is None + and not self.sessions + ): self._plugin_config_needs_reinit = True + def shutdown(self) -> None: + """Close active sessions and the process-lifetime plugin activation.""" + for session_id in list(self.sessions): + self.close_session({"session_id": session_id, "reason": "runtime_shutdown"}) + if self._plugin_config_initialized: + try: + self._clear_plugins_toml() + except Exception: + logger.debug("NeMo Relay plugin runtime shutdown failed", exc_info=True) + self._clear_atof() + if self._shutdown_registered: + atexit.unregister(self.shutdown) + self._shutdown_registered = False + def mark(self, name: str, kwargs: dict[str, Any]) -> None: state = self.ensure_session(kwargs) self.nemo_relay.scope.event( @@ -274,14 +341,14 @@ class _Runtime: def managed_llm_enabled(self) -> bool: return ( - self.settings.adaptive_enabled + (self.settings.adaptive_enabled or self._plugin_activation is not None) and callable(getattr(getattr(self.nemo_relay, "llm", None), "execute", None)) and callable(getattr(self.nemo_relay, "LLMRequest", None)) ) def managed_tool_enabled(self) -> bool: return ( - self.settings.adaptive_enabled + (self.settings.adaptive_enabled or self._plugin_activation is not None) and callable(getattr(getattr(self.nemo_relay, "tools", None), "execute", None)) ) @@ -402,6 +469,10 @@ class _Runtime: def register(ctx) -> None: + # Activate dynamic plugins before Hermes installs the managed execution + # boundaries that invoke their interceptors. + if _load_settings().dynamic_plugins: + _get_runtime() ctx.register_hook("on_session_start", on_session_start) ctx.register_hook("on_session_end", on_session_end) ctx.register_hook("on_session_finalize", on_session_finalize) @@ -648,6 +719,7 @@ def _load_settings() -> _Settings: return _Settings( plugins_toml_path=plugins_toml_path, plugins_config=plugins_config, + dynamic_plugins=_dynamic_plugin_specs(plugins_config), adaptive_enabled=adaptive_config is not None, adaptive_mode=_adaptive_mode(adaptive_config), atof_enabled=_env_bool("HERMES_NEMO_RELAY_ATOF_ENABLED"), @@ -664,6 +736,81 @@ def _load_settings() -> _Settings: ) +def _static_plugin_config(plugins_config: dict[str, Any]) -> dict[str, Any]: + """Return Relay's base plugin config without Hermes-owned host fields.""" + return {key: value for key, value in plugins_config.items() if key != "dynamic_plugins"} + + +def _dynamic_plugin_specs(plugins_config: dict[str, Any] | None) -> list[dict[str, Any]]: + if not isinstance(plugins_config, dict): + return [] + raw_specs = plugins_config.get("dynamic_plugins") + if raw_specs is None: + return [] + if not isinstance(raw_specs, list): + logger.warning( + "Ignoring invalid NeMo Relay dynamic_plugins config: expected an array of plugin specs" + ) + return [] + + specs: list[dict[str, Any]] = [] + for index, raw_spec in enumerate(raw_specs): + if not isinstance(raw_spec, dict): + logger.warning( + "Ignoring invalid NeMo Relay dynamic_plugins[%d]: expected an object", index + ) + continue + plugin_id = raw_spec.get("plugin_id") + kind = raw_spec.get("kind") + manifest_ref = raw_spec.get("manifest_ref") + config = raw_spec.get("config", {}) + environment_ref = raw_spec.get("environment_ref") + if not isinstance(plugin_id, str) or not plugin_id.strip(): + logger.warning( + "Ignoring invalid NeMo Relay dynamic_plugins[%d]: plugin_id is required", index + ) + continue + if kind not in {"rust_dynamic", "worker"}: + logger.warning( + "Ignoring invalid NeMo Relay dynamic_plugins[%d]: kind must be rust_dynamic or worker", + index, + ) + continue + if not isinstance(manifest_ref, str) or not manifest_ref.strip(): + logger.warning( + "Ignoring invalid NeMo Relay dynamic_plugins[%d]: manifest_ref is required", index + ) + continue + if not isinstance(config, dict): + logger.warning( + "Ignoring invalid NeMo Relay dynamic_plugins[%d]: config must be an object", index + ) + continue + if environment_ref is not None and not isinstance(environment_ref, str): + logger.warning( + "Ignoring invalid NeMo Relay dynamic_plugins[%d]: environment_ref must be a string", + index, + ) + continue + spec: dict[str, Any] = { + "plugin_id": plugin_id, + "kind": kind, + "manifest_ref": manifest_ref, + "config": config, + } + if environment_ref is not None: + spec["environment_ref"] = environment_ref + specs.append(spec) + return specs + + +def _flush_relay_subscribers(nemo_relay: Any) -> None: + subscribers = getattr(nemo_relay, "subscribers", None) + flush = getattr(subscribers, "flush", None) + if callable(flush): + flush() + + def _load_plugins_config(path: str) -> dict[str, Any] | None: if not path: return None @@ -959,4 +1106,6 @@ def _resolve_awaitable(value: Any) -> Any: def reset_for_tests() -> None: global _RUNTIME with _LOCK: + if isinstance(_RUNTIME, _Runtime): + _RUNTIME.shutdown() _RUNTIME = None diff --git a/pyproject.toml b/pyproject.toml index 963210075e0a..564f9572d7bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -205,7 +205,7 @@ vision = [] # extra that exposes a Starlette-backed server surface so pip/uv can't resolve # a vulnerable pre-1.0.1 transitive. Bump in lockstep with uv.lock. mcp = ["mcp==1.26.0", "starlette==1.0.1"] # starlette: CVE-2026-48710 -nemo-relay = ["nemo-relay==0.3"] +nemo-relay = ["nemo-relay>=0.5,<0.7"] homeassistant = ["aiohttp==3.14.1"] sms = ["aiohttp==3.14.1"] teams = ["microsoft-teams-apps==2.0.13.4", "aiohttp==3.14.1"] # aiohttp 3.14.1: CVE-2026-34993(RCE)/47265 + 34513/34518/34519/34520/34525 diff --git a/tests/plugins/test_nemo_relay_plugin.py b/tests/plugins/test_nemo_relay_plugin.py index 1e72520d2991..263e98c74b1c 100644 --- a/tests/plugins/test_nemo_relay_plugin.py +++ b/tests/plugins/test_nemo_relay_plugin.py @@ -41,7 +41,12 @@ class _FakeNemoRelay: call_end=self._tool_call_end, execute=self._tool_execute, ) - self.plugin = SimpleNamespace(initialize=self._plugin_initialize, clear=self._plugin_clear) + self.plugin = SimpleNamespace( + initialize=self._plugin_initialize, + clear=self._plugin_clear, + activate_dynamic_plugins=self._plugin_activate_dynamic, + ) + self.subscribers = SimpleNamespace(flush=self._flush_subscribers) self.LLMRequest = _FakeLLMRequest self.AtofExporterConfig = _FakeAtofExporterConfig self.AtofExporterMode = SimpleNamespace(Append="append", Overwrite="overwrite") @@ -100,6 +105,22 @@ class _FakeNemoRelay: async def _plugin_clear(self): self.events.append(("plugin.clear",)) + async def _plugin_activate_dynamic(self, config, dynamic_plugins): + self.events.append(("plugin.activate_dynamic", config, dynamic_plugins)) + return _FakePluginActivation(self.events) + + def _flush_subscribers(self): + self.events.append(("subscribers.flush",)) + + +class _FakePluginActivation: + def __init__(self, events): + self.events = events + self.report = {"diagnostics": []} + + async def close(self): + self.events.append(("plugin.activation.close",)) + class _FakeLLMRequest: def __init__(self, headers, content): @@ -143,6 +164,7 @@ class _FakeAtifExporter: return True def export_json(self): + self.events.append(("atif.export", self.session_id)) return json.dumps({"session_id": self.session_id, "agent_name": self.agent_name}) @@ -181,6 +203,26 @@ mode = "observe_only" monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) +def _enable_dynamic_plugin(tmp_path, monkeypatch) -> Path: + plugins_toml = tmp_path / "plugins.toml" + plugins_toml.write_text( + f""" +version = 1 + +[[dynamic_plugins]] +plugin_id = "fixture" +kind = "rust_dynamic" +manifest_ref = "{(tmp_path / "fixture" / "relay-plugin.toml").as_posix()}" + +[dynamic_plugins.config] +mode = "test" +""", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) + return plugins_toml + + def test_manifest_fields(): data = yaml.safe_load((PLUGIN_DIR / "plugin.yaml").read_text()) assert data["name"] == "nemo_relay" @@ -508,6 +550,151 @@ enabled = true assert event_names.count("plugin.clear") == 1 +def test_nemo_relay_plugin_activates_and_owns_dynamic_plugins(tmp_path, monkeypatch): + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") + monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif")) + + plugin.on_session_start(session_id="s1") + runtime = plugin._get_runtime() + assert runtime is not None + assert runtime._plugin_activation is not None + llm_result = plugin.on_llm_execution_middleware( + session_id="s1", + provider="openai", + model="fixture", + request={"messages": []}, + next_call=lambda request: {"request": request}, + ) + tool_result = plugin.on_tool_execution_middleware( + session_id="s1", + tool_name="fixture-tool", + args={"value": 1}, + next_call=lambda args: {"args": args}, + ) + assert llm_result["request"]["intercepted"] is True + assert tool_result["args"]["intercepted"] is True + plugin.on_session_finalize(session_id="s1", reason="shutdown") + assert runtime._plugin_activation is not None + assert not any(event[0] == "plugin.activation.close" for event in fake.events) + plugin.on_session_start(session_id="s2") + plugin.on_session_finalize(session_id="s2", reason="shutdown") + assert sum(event[0] == "plugin.activate_dynamic" for event in fake.events) == 1 + + activation = next(event for event in fake.events if event[0] == "plugin.activate_dynamic") + assert "dynamic_plugins" not in activation[1] + assert activation[2] == [ + { + "plugin_id": "fixture", + "kind": "rust_dynamic", + "manifest_ref": str(tmp_path / "fixture" / "relay-plugin.toml"), + "config": {"mode": "test"}, + } + ] + event_names = [event[0] for event in fake.events] + assert "plugin.clear" not in event_names + assert event_names.index("subscribers.flush") < event_names.index("atif.export") + assert event_names.index("atif.export") < event_names.index("atif.deregister") + + runtime.shutdown() + event_names = [event[0] for event in fake.events] + assert event_names.index("atif.deregister") < event_names.index("plugin.activation.close") + + +def test_nemo_relay_plugin_activates_before_registering_managed_middleware(tmp_path, monkeypatch): + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + + class _Context: + def register_hook(self, name, callback): + del name, callback + + def register_middleware(self, name, callback): + del callback + fake.events.append(("hermes.register_middleware", name)) + + plugin.register(_Context()) + + event_names = [event[0] for event in fake.events] + assert event_names.index("plugin.activate_dynamic") < event_names.index( + "hermes.register_middleware" + ) + runtime = plugin._get_runtime() + assert runtime is not None + runtime.shutdown() + + +def test_nemo_relay_plugin_degrades_to_static_config_on_relay_0_5( + tmp_path, monkeypatch, caplog +): + fake = _FakeNemoRelay() + delattr(fake.plugin, "activate_dynamic_plugins") + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + + with caplog.at_level("WARNING"): + plugin.on_session_start(session_id="s1") + + initialize = next(event for event in fake.events if event[0] == "plugin.initialize") + assert "dynamic_plugins" not in initialize[1] + assert not any(event[0] == "plugin.activate_dynamic" for event in fake.events) + assert "require nemo-relay>=0.6,<0.7" in caplog.text + + +def test_nemo_relay_plugin_ignores_invalid_dynamic_specs(tmp_path, monkeypatch, caplog): + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + plugins_toml = tmp_path / "plugins.toml" + plugins_toml.write_text( + """ +version = 1 +dynamic_plugins = [{ kind = "rust_dynamic", manifest_ref = "missing-id" }] +""", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) + + with caplog.at_level("WARNING"): + plugin.on_session_start(session_id="s1") + + assert not any(event[0] == "plugin.activate_dynamic" for event in fake.events) + assert "plugin_id is required" in caplog.text + + +def test_nemo_relay_plugin_registers_shutdown_after_dynamic_retry(tmp_path, monkeypatch): + fake = _FakeNemoRelay() + activation_attempts = 0 + + async def _flaky_activate(config, dynamic_plugins): + nonlocal activation_attempts + activation_attempts += 1 + fake.events.append( + ("plugin.activate_dynamic.attempt", activation_attempts, config, dynamic_plugins) + ) + if activation_attempts == 1: + raise RuntimeError("temporary activation failure") + return _FakePluginActivation(fake.events) + + fake.plugin.activate_dynamic_plugins = _flaky_activate + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + + plugin.on_session_start(session_id="s1") + plugin.on_session_finalize(session_id="s1") + plugin.on_session_start(session_id="s2") + + runtime = plugin._get_runtime() + assert runtime is not None + assert runtime._plugin_activation is not None + assert runtime._shutdown_registered is True + assert activation_attempts == 2 + runtime.shutdown() + assert any(event[0] == "plugin.activation.close" for event in fake.events) + + def test_nemo_relay_plugin_keeps_plugins_toml_active_while_other_sessions_remain(tmp_path, monkeypatch): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake)