mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-20 15:33:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #59846 from bbednarski9/bbednarski/nemo-relay-upgrade

Browse source 
feat(nemo-relay): nemo-relay observability version upgrade to support dynamic plugin activation
This commit is contained in:
Jeffrey Quesnelle 2026-07-14 13:15:49 -04:00 • committed by GitHub
commit 77d5b2d573
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 859 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

77
plugins/observability/nemo_relay/README.md
View file

@ -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 <plugin-id> --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

332
plugins/observability/nemo_relay/__init__.py
View file

@ -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