diff --git a/plugins/observability/nemo_relay/README.md b/plugins/observability/nemo_relay/README.md index ddc27e491103..a1b90c4da4fa 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,<1.0" 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.x distribution beginning with 0.5: ```bash -pip install "nemo-relay==0.3" +pip install "nemo-relay>=0.5,<1.0" ``` ## Export Configuration @@ -189,16 +190,75 @@ 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 + +Hermes feature-detects the dynamic-plugin activation API available in NeMo Relay +0.6 and later. Configure native or worker plugins with Hermes-owned +`[[dynamic_plugins]]` entries that match the Python binding's activation-spec +fields: + +```toml +[[dynamic_plugins]] +plugin_id = "example-plugin" +kind = "rust_dynamic" +manifest_ref = "./example-plugin/relay-plugin.toml" + +[dynamic_plugins.config] +mode = "enabled" +``` + +For a worker plugin, also provide the lifecycle-managed `environment_ref`: + +```toml +[[dynamic_plugins]] +plugin_id = "example-worker" +kind = "worker" +manifest_ref = "./example-worker/relay-plugin.toml" +environment_ref = "/absolute/path/from-nemo-relay-plugins-inspect" + +[dynamic_plugins.config] +mode = "enabled" +``` + +Provision the worker first with `nemo-relay plugins add`, then copy +`data.source.environment_ref` from the JSON output of +`nemo-relay plugins inspect --json`. Relay rejects arbitrary Python +environments at activation time. + +Relative `manifest_ref` and `environment_ref` values resolve relative to the +physical `plugins.toml` file. + +Relay's canonical gateway `[[plugins.dynamic]]` records are not interchangeable +with this Hermes-owned section. The gateway combines those records with +separate lifecycle state for enablement, trust policy, and worker environments; +the Python binding does not yet expose that resolver. Hermes rejects +`[[plugins.dynamic]]` with an actionable diagnostic instead of silently +ignoring it or bypassing lifecycle policy. Use `[[dynamic_plugins]]` until Relay +exposes shared file-and-lifecycle resolution to embedding hosts. + +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 plugin configuration is present with a binding that lacks the +activation API, 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.x +distribution beginning with 0.5 and a local Ollama model served through the +OpenAI-compatible API. ```bash -pip install "nemo-relay==0.3" +pip install "nemo-relay>=0.5,<1.0" export HERMES_HOME=/tmp/hermes-nemo-relay-docs/hermes-home mkdir -p "$HERMES_HOME" @@ -444,9 +504,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.x release range beginning with 0.5. ```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..56c6140ca58c 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 @@ -19,6 +20,11 @@ logger = logging.getLogger(__name__) _INIT_FAILED = object() _LOCK = threading.RLock() _RUNTIME: "_Runtime | object | None" = None +_RELAY_LLM_SURFACE_BY_API_MODE = { + "anthropic_messages": "anthropic.messages", + "chat_completions": "openai.chat_completions", + "codex_responses": "openai.responses", +} @dataclass @@ -44,6 +50,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 +74,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,29 +85,88 @@ 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 a binding that exposes " + "plugin.activate_dynamic_plugins (available in NeMo Relay 0.6+). " + "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()) - finally: - self._plugin_config_initialized = False - self._plugin_config_needs_reinit = bool(self.settings.plugins_config) + failures: list[str] = [] + if self._plugin_activation is not None: + activation = self._plugin_activation + try: + _flush_relay_subscribers(self.nemo_relay) + except Exception as exc: + failures.append(f"subscriber flush failed: {exc}") + + close = getattr(activation, "close", None) + if callable(close): + try: + _resolve_awaitable(close()) + except Exception as exc: + failures.append(f"dynamic plugin activation close failed: {exc}") + finally: + # Retain the owned activation through the complete close + # attempt. The binding transitions it to a terminal state + # before its awaitable resolves, including error results. + self._plugin_activation = None + self._plugin_config_initialized = False + self._plugin_config_needs_reinit = bool(self.settings.plugins_config) + else: + failures.append("dynamic plugin activation has no close method") + else: + try: + plugin_mod = getattr(self.nemo_relay, "plugin", None) + clear = getattr(plugin_mod, "clear", None) + if callable(clear): + _resolve_awaitable(clear()) + except Exception as exc: + failures.append(f"static plugin configuration clear failed: {exc}") + finally: + self._plugin_config_initialized = False + self._plugin_config_needs_reinit = bool(self.settings.plugins_config) + + if failures: + raise RuntimeError("; ".join(failures)) def _activate_direct_fallbacks(self) -> None: self._plugin_config_needs_reinit = False @@ -221,24 +289,69 @@ class _Runtime: state = self.sessions.pop(session_id, None) if state is None: return + failures: list[str] = [] if state.handle is not None: try: self.nemo_relay.scope.pop(state.handle, output=_jsonable(kwargs)) - except Exception: - logger.debug("NeMo Relay session pop failed", exc_info=True) - self.export_atif(state) + except Exception as exc: + failures.append(f"session scope pop failed: {exc}") + try: + _flush_relay_subscribers(self.nemo_relay) + except Exception as exc: + failures.append(f"subscriber flush failed: {exc}") + try: + self.export_atif(state) + except Exception as exc: + failures.append(f"ATIF export failed: {exc}") 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: + except Exception as exc: + failures.append(f"ATIF deregister failed: {exc}") + 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: + except Exception as exc: + failures.append(f"plugin configuration clear failed: {exc}") + elif ( + self.settings.plugins_config + and self._plugin_activation is None + and not self.sessions + ): self._plugin_config_needs_reinit = True + if failures: + logger.warning( + "NeMo Relay session %s teardown completed with errors: %s", + session_id, + "; ".join(failures), + ) + + def shutdown(self) -> None: + """Close active sessions and the process-lifetime plugin activation.""" + failures: list[str] = [] + for session_id in list(self.sessions): + try: + self.close_session({"session_id": session_id, "reason": "runtime_shutdown"}) + except Exception as exc: + failures.append(f"session {session_id} close failed: {exc}") + if self._plugin_config_initialized: + try: + self._clear_plugins_toml() + except Exception as exc: + failures.append(f"plugin runtime close failed: {exc}") + self._clear_atof() + if self._shutdown_registered and self._plugin_activation is None: + atexit.unregister(self.shutdown) + self._shutdown_registered = False + if failures: + logger.warning( + "NeMo Relay runtime shutdown completed with errors: %s", + "; ".join(failures), + ) def mark(self, name: str, kwargs: dict[str, Any]) -> None: state = self.ensure_session(kwargs) @@ -274,14 +387,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)) ) @@ -291,6 +404,8 @@ class _Runtime: normalize_payload: Callable[[Any], Any], shape_response: Callable[[Any], Any], make_managed_execute: Callable[[Callable[[Any], Any]], Any], + *, + preserve_raw_response: bool, ) -> Any: # NeMo Relay's native managed execution may wrap a failing callback as an # internal runtime error, hiding the real downstream provider/tool @@ -298,7 +413,7 @@ class _Runtime: # execution so Hermes retry classification still sees it. The LLM and tool # paths share this scaffolding; they differ only in payload normalization, # response shaping, and the Relay call itself. - raw_response: dict[str, Any] = {"set": False, "value": None} + raw_response: dict[str, Any] = {"set": False, "value": None, "normalized": None} callback_error: Exception | None = None downstream_error: BaseException | None = None @@ -312,7 +427,8 @@ class _Runtime: raise raw_response["set"] = True raw_response["value"] = raw - return shape_response(raw) + raw_response["normalized"] = shape_response(raw) + return raw_response["normalized"] try: managed_result = _resolve_awaitable(make_managed_execute(_impl)) @@ -320,7 +436,13 @@ class _Runtime: if downstream_error is not None and _is_relay_wrapped_callback_error(exc, callback_error): raise downstream_error raise - return raw_response["value"] if raw_response["set"] else managed_result + if ( + preserve_raw_response + and raw_response["set"] + and _json_semantically_equal(managed_result, raw_response["normalized"]) + ): + return raw_response["value"] + return managed_result def execute_llm(self, kwargs: dict[str, Any]) -> Any: state = self.ensure_session(kwargs) @@ -337,7 +459,7 @@ class _Runtime: def _make_managed(impl: Callable[[Any], Any]) -> Any: async def _managed_execute() -> Any: result = self.nemo_relay.llm.execute( - str(kwargs.get("provider") or "llm"), + _relay_llm_surface(kwargs), request, impl, handle=state.handle, @@ -359,7 +481,7 @@ class _Runtime: return _managed_execute() return self._run_managed_with_downstream_preservation( - next_call, _normalize, _llm_response_payload, _make_managed + next_call, _normalize, _llm_response_payload, _make_managed, preserve_raw_response=True ) def execute_tool(self, kwargs: dict[str, Any]) -> Any: @@ -397,11 +519,15 @@ class _Runtime: return _managed_execute() return self._run_managed_with_downstream_preservation( - next_call, _normalize, _jsonable, _make_managed + next_call, _normalize, _jsonable, _make_managed, preserve_raw_response=False ) 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 +774,7 @@ def _load_settings() -> _Settings: return _Settings( plugins_toml_path=plugins_toml_path, plugins_config=plugins_config, + dynamic_plugins=_dynamic_plugin_specs(plugins_config, plugins_toml_path), adaptive_enabled=adaptive_config is not None, adaptive_mode=_adaptive_mode(adaptive_config), atof_enabled=_env_bool("HERMES_NEMO_RELAY_ATOF_ENABLED"), @@ -664,6 +791,136 @@ def _load_settings() -> _Settings: ) +def _static_plugin_config(plugins_config: dict[str, Any]) -> dict[str, Any]: + """Return Relay's base config without embedding- or gateway-host fields.""" + return { + key: value + for key, value in plugins_config.items() + if key not in {"dynamic_plugins", "plugins"} + } + + +def _dynamic_plugin_specs( + plugins_config: dict[str, Any] | None, + plugins_toml_path: str = "", +) -> list[dict[str, Any]]: + if not isinstance(plugins_config, dict): + return [] + + raw_specs = plugins_config.get("dynamic_plugins") + plugins_section = plugins_config.get("plugins") + if plugins_section is not None: + if not isinstance(plugins_section, dict): + logger.error( + "Invalid NeMo Relay plugins config: expected [plugins] to be an object; " + "no dynamic plugins will be activated. Continuing with static " + "observability only." + ) + return [] + if plugins_section: + logger.error( + "Hermes cannot activate Relay gateway [[plugins.dynamic]] records because " + "the Python binding does not expose the CLI lifecycle resolver for " + "enablement, trust policy, and worker environments. Use Hermes-owned " + "[[dynamic_plugins]] activation specs instead; no dynamic plugins will be " + "activated. Continuing with static observability only." + ) + return [] + 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]] = [] + invalid = False + for index, raw_spec in enumerate(raw_specs): + if not isinstance(raw_spec, dict): + logger.warning( + "Invalid NeMo Relay dynamic_plugins[%d]: expected an object", index + ) + invalid = True + 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( + "Invalid NeMo Relay dynamic_plugins[%d]: plugin_id is required", index + ) + invalid = True + continue + if kind not in {"rust_dynamic", "worker"}: + logger.warning( + "Invalid NeMo Relay dynamic_plugins[%d]: kind must be rust_dynamic or worker", + index, + ) + invalid = True + continue + if not isinstance(manifest_ref, str) or not manifest_ref.strip(): + logger.warning( + "Invalid NeMo Relay dynamic_plugins[%d]: manifest_ref is required", index + ) + invalid = True + continue + if not isinstance(config, dict): + logger.warning( + "Invalid NeMo Relay dynamic_plugins[%d]: config must be an object", index + ) + invalid = True + continue + if environment_ref is not None and ( + not isinstance(environment_ref, str) or not environment_ref.strip() + ): + logger.warning( + "Invalid NeMo Relay dynamic_plugins[%d]: environment_ref must be a " + "non-empty string", + index, + ) + invalid = True + continue + spec: dict[str, Any] = { + "plugin_id": plugin_id.strip(), + "kind": kind, + "manifest_ref": _config_relative_path(manifest_ref.strip(), plugins_toml_path), + "config": config, + } + if environment_ref is not None: + spec["environment_ref"] = _config_relative_path( + environment_ref.strip(), plugins_toml_path + ) + specs.append(spec) + if invalid: + logger.error( + "NeMo Relay dynamic plugin configuration is invalid; no dynamic plugins " + "will be activated. Continuing with static observability only." + ) + return [] + return specs + + +def _config_relative_path(value: str, plugins_toml_path: str) -> str: + """Resolve a plugin path relative to its physical ``plugins.toml`` file.""" + path = Path(value) + if path.is_absolute(): + return str(path) + config_path = Path(plugins_toml_path) if plugins_toml_path else Path.cwd() / "plugins.toml" + if not config_path.is_absolute(): + config_path = Path.cwd() / config_path + return os.path.abspath(config_path.parent / path) + + +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 @@ -776,6 +1033,14 @@ def _tool_key(kwargs: dict[str, Any]) -> str: ) +def _relay_llm_surface(kwargs: dict[str, Any]) -> str: + api_mode = str(kwargs.get("api_mode") or "").strip().lower() + return _RELAY_LLM_SURFACE_BY_API_MODE.get( + api_mode, + str(kwargs.get("provider") or "llm"), + ) + + def _metadata(kwargs: dict[str, Any]) -> dict[str, Any]: keys = ( "telemetry_schema_version", @@ -835,6 +1100,15 @@ def _jsonable(value: Any) -> Any: return str(value) +def _json_semantically_equal(left: Any, right: Any) -> bool: + """Compare JSON-compatible values without conflating booleans and numbers.""" + try: + options = {"ensure_ascii": False, "sort_keys": True, "separators": (",", ":")} + return json.dumps(_jsonable(left), **options) == json.dumps(_jsonable(right), **options) + except (TypeError, ValueError): + return False + + def _value(obj: Any, key: str, default: Any = None) -> Any: if isinstance(obj, dict): return obj.get(key, default) @@ -959,4 +1233,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 851473b13b3f..6f836ca3233b 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,<1.0"] 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..61958194e585 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,432 @@ 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_rejects_gateway_dynamic_config_with_actionable_diagnostic( + tmp_path, monkeypatch, caplog +): + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + plugins_toml = tmp_path / "plugins.toml" + plugins_toml.write_text( + """ +version = 1 + +[[plugins.dynamic]] +manifest = "plugins/fixture/relay-plugin.toml" + +[plugins.dynamic.config] +mode = "test" +""", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) + + with caplog.at_level("ERROR"): + plugin.on_session_start(session_id="s1") + + assert not any(event[0] == "plugin.activate_dynamic" for event in fake.events) + initialize = next(event for event in fake.events if event[0] == "plugin.initialize") + assert initialize[1] == {"version": 1} + assert "does not expose the CLI lifecycle resolver" in caplog.text + assert "Use Hermes-owned [[dynamic_plugins]]" in caplog.text + + +def test_nemo_relay_explicit_dynamic_paths_resolve_from_plugins_toml(tmp_path, monkeypatch): + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + config_dir = tmp_path / "config" + config_dir.mkdir() + plugins_toml = config_dir / "plugins.toml" + plugins_toml.write_text( + """ +version = 1 + +[[dynamic_plugins]] +plugin_id = "worker-fixture" +kind = "worker" +manifest_ref = "../plugins/worker/relay-plugin.toml" +environment_ref = "../environments/worker-fixture" +""", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) + + plugin.on_session_start(session_id="s1") + + activation = next(event for event in fake.events if event[0] == "plugin.activate_dynamic") + assert activation[2] == [ + { + "plugin_id": "worker-fixture", + "kind": "worker", + "manifest_ref": str(tmp_path / "plugins" / "worker" / "relay-plugin.toml"), + "environment_ref": str(tmp_path / "environments" / "worker-fixture"), + "config": {}, + } + ] + runtime = plugin._get_runtime() + assert runtime is not None + runtime.shutdown() + + +@pytest.mark.parametrize( + ("provider", "api_mode", "expected_surface", "should_rewrite"), + [ + ("custom", "chat_completions", "openai.chat_completions", True), + ("openai-codex", "codex_responses", "openai.responses", True), + ("anthropic", "anthropic_messages", "anthropic.messages", True), + ("custom", "anthropic_messages", "anthropic.messages", True), + ("bedrock", "bedrock_converse", "bedrock", False), + ], +) +def test_nemo_relay_managed_llm_uses_wire_protocol_for_interceptor_dispatch( + tmp_path, + monkeypatch, + provider, + api_mode, + expected_surface, + should_rewrite, +): + fake = _FakeNemoRelay() + supported_surfaces = { + "anthropic.messages", + "openai.chat_completions", + "openai.responses", + } + + def execute(name, request, func, **kwargs): + fake.events.append(("llm.execute.start", name, request.content, kwargs)) + content = dict(request.content) + if name in supported_surfaces: + content["rewritten_for"] = name + result = func(_FakeLLMRequest(request.headers, content)) + fake.events.append(("llm.execute.end", name, result, kwargs)) + return result + + fake.llm.execute = execute + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + + result = plugin.on_llm_execution_middleware( + session_id="s1", + provider=provider, + api_mode=api_mode, + model="fixture", + request={"messages": [{"role": "user", "content": "hi"}]}, + next_call=lambda request: request, + ) + + execute_start = next( + event for event in fake.events if event[0] == "llm.execute.start" + ) + assert execute_start[1] == expected_surface + assert execute_start[3]["metadata"]["provider"] == provider + assert execute_start[3]["metadata"]["api_mode"] == api_mode + if should_rewrite: + assert result["rewritten_for"] == expected_surface + else: + assert "rewritten_for" not in result + + +def test_nemo_relay_managed_llm_returns_post_next_interceptor_result(tmp_path, monkeypatch): + fake = _FakeNemoRelay() + raw_response = SimpleNamespace( + model="fixture", + choices=[ + SimpleNamespace( + message=SimpleNamespace(role="assistant", content="raw", tool_calls=[]), + finish_reason="stop", + ) + ], + usage=None, + ) + + def execute(name, request, func, **kwargs): + del name, kwargs + normalized = func(_FakeLLMRequest(request.headers, request.content)) + return {**normalized, "post_next_interceptor": True} + + fake.llm.execute = execute + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + + result = plugin.on_llm_execution_middleware( + session_id="s1", + provider="openai", + api_mode="chat_completions", + model="fixture", + request={"messages": []}, + next_call=lambda request: raw_response, + ) + + assert result["post_next_interceptor"] is True + assert result["assistant_message"]["content"] == "raw" + assert result is not raw_response + + +def test_nemo_relay_managed_tool_returns_post_interceptor_result(tmp_path, monkeypatch): + fake = _FakeNemoRelay() + + def execute(name, args, func, **kwargs): + fake.events.append(("tool.execute.start", name, args, kwargs)) + raw = func({"intercepted": True, **args}) + result = {"compressed": True, "raw": raw} + fake.events.append(("tool.execute.end", name, result, kwargs)) + return result + + fake.tools.execute = execute + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + + result = plugin.on_tool_execution_middleware( + session_id="s1", + tool_name="fixture-tool", + args={"value": 1}, + next_call=lambda args: {"tool_output": args}, + ) + + assert result == { + "compressed": True, + "raw": {"tool_output": {"intercepted": True, "value": 1}}, + } + + +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 "available in NeMo Relay 0.6+" in caplog.text + + +def test_nemo_relay_plugin_rejects_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_rejects_entire_mixed_valid_invalid_dynamic_request( + tmp_path, monkeypatch, caplog +): + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + plugins_toml = tmp_path / "plugins.toml" + plugins_toml.write_text( + f""" +version = 1 + +[[dynamic_plugins]] +plugin_id = "valid-fixture" +kind = "rust_dynamic" +manifest_ref = "{(tmp_path / "valid" / "relay-plugin.toml").as_posix()}" + +[[dynamic_plugins]] +kind = "worker" +manifest_ref = "{(tmp_path / "invalid" / "relay-plugin.toml").as_posix()}" +""", + 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) + initialize = next(event for event in fake.events if event[0] == "plugin.initialize") + assert "dynamic_plugins" not in initialize[1] + assert "no dynamic plugins will be activated" 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_attempts_activation_close_after_subscriber_flush_failure( + tmp_path, monkeypatch, caplog +): + fake = _FakeNemoRelay() + + def _failing_flush(): + fake.events.append(("subscribers.flush.failed",)) + raise RuntimeError("flush boom") + + fake.subscribers.flush = _failing_flush + plugin = _fresh_plugin(monkeypatch, fake) + _enable_dynamic_plugin(tmp_path, monkeypatch) + plugin.on_session_start(session_id="s1") + runtime = plugin._get_runtime() + assert runtime is not None + + with caplog.at_level("WARNING"): + runtime.shutdown() + + event_names = [event[0] for event in fake.events] + assert event_names.count("subscribers.flush.failed") == 2 + flush_indices = [ + index for index, name in enumerate(event_names) if name == "subscribers.flush.failed" + ] + assert max(flush_indices) < event_names.index("plugin.activation.close") + assert runtime._plugin_activation is None + assert "subscriber flush failed: flush boom" in caplog.text + + +def test_nemo_relay_plugin_continues_shutdown_after_atif_export_failure( + tmp_path, monkeypatch, caplog +): + fake = _FakeNemoRelay() + + class _FailingAtifExporter(_FakeAtifExporter): + def export_json(self): + self.events.append(("atif.export.failed", self.session_id)) + raise OSError("disk full") + + fake.AtifExporter = lambda session_id, agent_name, agent_version, **kwargs: ( + _FailingAtifExporter(fake.events, session_id, agent_name, agent_version, kwargs) + ) + 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 + + with caplog.at_level("WARNING"): + runtime.shutdown() + + event_names = [event[0] for event in fake.events] + assert event_names.index("atif.export.failed") < event_names.index("atif.deregister") + assert event_names.index("atif.deregister") < event_names.index("plugin.activation.close") + assert runtime._plugin_activation is None + assert "ATIF export failed: disk full" in caplog.text + + def test_nemo_relay_plugin_keeps_plugins_toml_active_while_other_sessions_remain(tmp_path, monkeypatch): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) @@ -764,15 +1232,16 @@ mode = "observe_only" ), finish_reason="tool_calls", ) + raw_response = SimpleNamespace( + id="resp-1", + model="demo-model", + choices=[raw_choice], + usage=SimpleNamespace(prompt_tokens=3, completion_tokens=5, total_tokens=8), + ) def next_call(request): seen_request.update(request) - return SimpleNamespace( - id="resp-1", - model="demo-model", - choices=[raw_choice], - usage=SimpleNamespace(prompt_tokens=3, completion_tokens=5, total_tokens=8), - ) + return raw_response response = plugin.on_llm_execution_middleware( session_id="s1", @@ -786,6 +1255,7 @@ mode = "observe_only" next_call=next_call, ) + assert response is raw_response assert response.model == "demo-model" assert response.choices == [raw_choice] assert seen_request["intercepted"] is True diff --git a/tests/test_project_metadata.py b/tests/test_project_metadata.py index f2f4887d6091..8c0836e9059e 100644 --- a/tests/test_project_metadata.py +++ b/tests/test_project_metadata.py @@ -222,10 +222,10 @@ def test_feishu_extra_includes_qrcode_for_qr_login(): assert any(dep.startswith("qrcode") for dep in feishu_extra) -def test_nemo_relay_extra_uses_official_0_3_distribution(): +def test_nemo_relay_extra_uses_supported_official_distribution_range(): optional_dependencies = _load_optional_dependencies() - assert optional_dependencies["nemo-relay"] == ["nemo-relay==0.3"] + assert optional_dependencies["nemo-relay"] == ["nemo-relay>=0.5,<1.0"] assert not any( spec == "hermes-agent[nemo-relay]" for spec in optional_dependencies["all"] diff --git a/uv.lock b/uv.lock index 21bd0827c77e..9bd8d16a6993 100644 --- a/uv.lock +++ b/uv.lock @@ -1804,7 +1804,7 @@ requires-dist = [ { name = "microsoft-teams-apps", marker = "extra == 'teams'", specifier = "==2.0.13.4" }, { name = "mistralai", marker = "extra == 'mistral'", specifier = "==2.4.8" }, { name = "modal", marker = "extra == 'modal'", specifier = "==1.3.4" }, - { name = "nemo-relay", marker = "extra == 'nemo-relay'", specifier = "==0.3" }, + { name = "nemo-relay", marker = "extra == 'nemo-relay'", specifier = ">=0.5,<1.0" }, { name = "numpy", marker = "extra == 'voice'", specifier = "==2.4.3" }, { name = "openai", specifier = "==2.24.0" }, { name = "packaging", specifier = "==26.0" }, @@ -2608,14 +2608,14 @@ wheels = [ [[package]] name = "nemo-relay" -version = "0.3.0" +version = "0.5.0" source = { registry = "https://pypi.org/simple" } wheels = [ - { url = "https://files.pythonhosted.org/packages/15/55/3b65643db2df02fb3838b3d251442e05b7306cbbc77e69884ae78743546f/nemo_relay-0.3.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:a6e44abe38bf6dda8f6a8b149cd9069a548186f24263ae8469d5a37cefc95209", size = 5282297, upload-time = "2026-05-29T22:32:36.315Z" }, - { url = "https://files.pythonhosted.org/packages/c7/ef/e2f4bc02f99f38706fdde0cdda6b6d8fddea9e6975da1b66377c35958b85/nemo_relay-0.3.0-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:99ba247121cd0d17b9c5e2a95beb42512243430773b93435848b8b4b9f9ca61f", size = 4722431, upload-time = "2026-05-29T22:32:38.088Z" }, - { url = "https://files.pythonhosted.org/packages/12/18/bedb5d57f206e2859cef11f12f568974a529acf382436523a37e095b2cc7/nemo_relay-0.3.0-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:20bb1e4ed87f179befc4db3af2e35f08be711378b322f3672222d80294f57be7", size = 5002734, upload-time = "2026-05-29T22:32:40.191Z" }, - { url = "https://files.pythonhosted.org/packages/26/ec/4b2e758a0a25397f2e97be889a11b7787d108658e0b60ddff5048b25adb8/nemo_relay-0.3.0-cp311-abi3-win_amd64.whl", hash = "sha256:e659c2772b35c0ea4897a91f6bf86b551ed403f6f41d0b60fd8151f5c78b83d0", size = 4947785, upload-time = "2026-05-29T22:32:41.784Z" }, - { url = "https://files.pythonhosted.org/packages/2b/91/45a3397559679d846ae023a82527c43b14631786b3a7c95e69c05e8bb553/nemo_relay-0.3.0-cp311-abi3-win_arm64.whl", hash = "sha256:9655514adb518e19caf7b3f6a08bbe82c1b24759c0a8021c3df949fd265bcef6", size = 4662147, upload-time = "2026-05-29T22:32:43.554Z" }, + { url = "https://files.pythonhosted.org/packages/aa/fc/6892859ffbfb60f8cc09a4181c8d407b574d49490e35cd563e7f0a61bf54/nemo_relay-0.5.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:203409e0f392fc4f40f52277347ed75ba9969b2af28ddb3b6ae53f789185a0ed", size = 7769721, upload-time = "2026-07-08T18:43:57.745Z" }, + { url = "https://files.pythonhosted.org/packages/dd/13/25c4fcf6fb979af07c9562238819183588a910c2ae09767603500d6948c2/nemo_relay-0.5.0-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:de61ad0644c5e8c300de814f6884c5de7ae51db14694f5057f938b4da54bfaaf", size = 7037083, upload-time = "2026-07-08T18:44:00.135Z" }, + { url = "https://files.pythonhosted.org/packages/49/97/bd4c77e975d50f7d09f8bfcb74d7e704e05a8b2205f555bf557f84edc5fe/nemo_relay-0.5.0-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:cb323569f104506f4f1a0e2104b22d74c9a0bbf3280ec4d58ad3d768de76fb15", size = 7430817, upload-time = "2026-07-08T18:44:02.165Z" }, + { url = "https://files.pythonhosted.org/packages/da/8b/5681f3430e6d25bf7419dbe576da14599e1cfadc083edb74e691a0c3a980/nemo_relay-0.5.0-cp311-abi3-win_amd64.whl", hash = "sha256:da1bd1e1dec79b7bda3984dd8c59b32ba97abf96c37e85cd1f38d208a29f42b9", size = 7359048, upload-time = "2026-07-08T18:44:04.169Z" }, + { url = "https://files.pythonhosted.org/packages/4e/f2/65c3ef0435e671c829063d4872897f9054f369f0aeacab12c5aefe486705/nemo_relay-0.5.0-cp311-abi3-win_arm64.whl", hash = "sha256:87462585f17b40ce82c54347acef5ace96ee11de3015f234e0b902f23b3793fb", size = 6934235, upload-time = "2026-07-08T18:44:06.078Z" }, ] [[package]]