From 5fbdd5a9e8e637cfbf49f3bffb0167d19aeccb4b Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 11:27:21 -0700 Subject: [PATCH 01/12] feat(hermes): use upstream Relay integration Signed-off-by: Zhongxuan Wang --- .../nemo_fabric_adapters/hermes/adapter.py | 220 +++++++++--------- python/src/nemo_fabric/streaming.py | 3 + tests/adapters/test_hermes_adapter.py | 180 +++++++------- tests/e2e/test_hermes_e2e.py | 64 +++-- tests/e2e/test_hermes_runtime.py | 5 +- tests/python/test_streaming.py | 9 +- 6 files changed, 257 insertions(+), 224 deletions(-) diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index 29d69b7a..648269cc 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -11,6 +11,7 @@ from __future__ import annotations import asyncio +import copy import inspect import json import logging @@ -35,6 +36,15 @@ "openai": "OPENAI_API_KEY", "openrouter": "OPENROUTER_API_KEY", } +HERMES_RELAY_ENV_NAMES = ( + "HERMES_NEMO_RELAY_PLUGINS_TOML", + "HERMES_NEMO_RELAY_ATIF_ENABLED", + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", + "HERMES_NEMO_RELAY_ATIF_FILENAME_TEMPLATE", + "HERMES_NEMO_RELAY_ATIF_AGENT_NAME", + "HERMES_NEMO_RELAY_ATIF_AGENT_VERSION", + "HERMES_NEMO_RELAY_ATIF_MODEL_NAME", +) def _api_key_env(model_config: dict[str, Any]) -> str: @@ -50,26 +60,6 @@ def _api_key_env(model_config: dict[str, Any]) -> str: return default -def _fabric_stream_sink_enabled(config: dict[str, Any] | None) -> bool: - if config is None: - return False - for component in config.get("components") or []: - if not isinstance(component, dict) or component.get("kind") != "observability": - continue - component_config = component.get("config") - if not isinstance(component_config, dict): - continue - atof = component_config.get("atof") - if not isinstance(atof, dict): - continue - if any( - isinstance(sink, dict) and sink.get("name") == "nemo-fabric-stream" - for sink in atof.get("sinks") or [] - ): - return True - return False - - def validate_hermes_telemetry_provider(payload: dict[str, Any]) -> None: providers = common_utils.telemetry_providers(payload) if any(provider != "relay" for provider in providers): @@ -153,6 +143,61 @@ def write_hermes_config( return config_path, config +def write_hermes_relay_plugin_config( + payload: dict[str, Any], +) -> tuple[Path, dict[str, Any]]: + """Stage Fabric's resolved Relay config for Hermes' bundled integration.""" + + plugin_config = common_utils.load_relay_plugin_config(payload) + hermes_plugin_config = copy.deepcopy(plugin_config) + for component in hermes_plugin_config.get("components", []): + if component.get("kind") != "observability": + continue + observability = component.get("config") + if isinstance(observability, dict): + # Hermes v0.20.0 manages ATIF through its plugin environment. Keep + # the complete Fabric config for artifact collection, but do not + # configure ATIF twice through the component TOML. + observability.pop("atif", None) + _, plugin_config_path = common_utils.write_relay_configs( + plugin_config=hermes_plugin_config + ) + if plugin_config_path is None: + raise RuntimeError("Hermes Relay plugin configuration was not generated") + return plugin_config_path, plugin_config + + +def hermes_relay_environment( + plugin_config: dict[str, Any], + plugin_config_path: Path, +) -> dict[str, str]: + """Translate Fabric's ATIF settings to Hermes' upstream plugin environment.""" + + environment = {"HERMES_NEMO_RELAY_PLUGINS_TOML": str(plugin_config_path)} + for component in plugin_config.get("components", []): + if component.get("kind") != "observability": + continue + observability = component.get("config") + if not isinstance(observability, dict): + continue + atif = observability.get("atif") + if not isinstance(atif, dict) or not atif.get("enabled"): + continue + environment["HERMES_NEMO_RELAY_ATIF_ENABLED"] = "1" + for config_key, environment_key in ( + ("output_directory", "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY"), + ("filename_template", "HERMES_NEMO_RELAY_ATIF_FILENAME_TEMPLATE"), + ("agent_name", "HERMES_NEMO_RELAY_ATIF_AGENT_NAME"), + ("agent_version", "HERMES_NEMO_RELAY_ATIF_AGENT_VERSION"), + ("model_name", "HERMES_NEMO_RELAY_ATIF_MODEL_NAME"), + ): + value = atif.get(config_key) + if value is not None: + environment[environment_key] = str(value) + break + return environment + + def hermes_mcp_server_config(server: dict[str, Any]) -> dict[str, Any]: transport = str(server.get("transport") or "").strip().lower() raw_target = server.get("url") @@ -231,13 +276,10 @@ def __init__(self) -> None: self._conversation_history: list[dict[str, Any]] | None = None self._session_db: Any = None self._agent: Any = None - self._invoke_hook: Any = None self._relay_plugin_config: dict[str, Any] | None = None - self._relay_context: Any = None - self._relay_context_entered = False - self._relay_session_pending = False - self._relay_finalize_hook_invoked = False - self._relay_model_name = "unknown" + self._relay_plugin_config_path: Path | None = None + self._previous_relay_environment: dict[str, str | None] = {} + self._applied_relay_environment: dict[str, str | None] = {} async def start(self, payload: dict[str, Any]) -> None: if self._started: @@ -247,8 +289,6 @@ async def start(self, payload: dict[str, Any]) -> None: ) try: - self._relay_session_pending = False - self._relay_finalize_hook_invoked = False validate_hermes_telemetry_provider(payload) self._settings = common_utils.settings_payload(payload) self._model_config = common_utils.selected_model_config(payload) @@ -270,14 +310,25 @@ async def start(self, payload: dict[str, Any]) -> None: relay_enabled = common_utils.relay_enabled(payload) if relay_enabled: - self._relay_plugin_config = common_utils.load_relay_plugin_config( - payload + ( + self._relay_plugin_config_path, + self._relay_plugin_config, + ) = write_hermes_relay_plugin_config(payload) + relay_environment = hermes_relay_environment( + self._relay_plugin_config, + self._relay_plugin_config_path, ) - from nemo_relay import plugin - - self._relay_context = plugin.plugin(self._relay_plugin_config) - await self._relay_context.__aenter__() - self._relay_context_entered = True + self._previous_relay_environment = { + name: os.environ.get(name) for name in HERMES_RELAY_ENV_NAMES + } + self._applied_relay_environment = {} + for name in HERMES_RELAY_ENV_NAMES: + value = relay_environment.get(name) + if value is None: + os.environ.pop(name, None) + else: + os.environ[name] = value + self._applied_relay_environment[name] = value self._hermes_config_path, self._hermes_config = write_hermes_config( payload, @@ -289,11 +340,9 @@ async def start(self, payload: dict[str, Any]) -> None: if not api_key: raise RuntimeError(f"{api_key_env} is required for Hermes mode") self._base_url = common_utils.get_base_url(self._model_config) - self._relay_model_name = common_utils.relay_model_name(payload) from hermes_cli.config import load_config from hermes_cli.plugins import discover_plugins - from hermes_cli.plugins import invoke_hook from hermes_state import SessionDB from run_agent import AIAgent @@ -303,10 +352,11 @@ async def start(self, payload: dict[str, Any]) -> None: # Hermes 0.12+ no longer discovers MCP tools as an import side effect # (#16856). Fabric is a Hermes host: discover after config.yaml exists # and before AIAgent resolves mcp-* toolsets. - # discover_mcp_tools uses a blocking 120s wait, wrapping it in + # discover_mcp_tools uses a blocking 120s wait, wrapping it in # asyncio.to_thread to avoid blocking the loop. if self._hermes_config.get("mcp_servers"): from tools.mcp_tool import discover_mcp_tools + await asyncio.to_thread(discover_mcp_tools) self._enabled_toolsets = resolve_hermes_toolsets( @@ -348,7 +398,6 @@ async def start(self, payload: dict[str, Any]) -> None: session_db=self._session_db, ) ) - self._invoke_hook = invoke_hook self._start_payload = payload self._started = True except BaseException: @@ -386,32 +435,10 @@ def invoke_turn() -> tuple[dict[str, Any], str]: conversation_history=self._conversation_history, ) - self._relay_session_pending = self._relay_plugin_config is not None - self._relay_finalize_hook_invoked = False - if _fabric_stream_sink_enabled(self._relay_plugin_config): - from nemo_relay import ScopeType, scope - - with scope.scope( - "nemo-fabric-invocation", - ScopeType.Agent, - metadata={ - "nemo_fabric_request_id": request.get("request_id"), - }, - ): - try: - result, adapter_stdout = invoke_turn() - finally: - # The Hermes plugin pushes its session below this correlation - # scope, so finalize that session before popping the parent. - self._finalize_relay_session() - else: - try: - result, adapter_stdout = invoke_turn() - finally: - # Hermes' Relay plugin materializes ATIF when its session-finalize - # hook runs. Finalize the telemetry session for each Fabric - # invocation while retaining the native AIAgent and SessionDB. - self._finalize_relay_session() + # Hermes' upstream Relay integration drives async Relay hooks from its + # synchronous agent loop. Run that loop outside this lifecycle server's + # event-loop thread so Hermes can own its Relay event loop. + result, adapter_stdout = await asyncio.to_thread(invoke_turn) messages = result.get("messages") or [] if isinstance(messages, list): self._conversation_history = messages @@ -439,51 +466,21 @@ def invoke_turn() -> tuple[dict[str, Any], str]: output["relay_runtime"] = { "enabled": True, "config_path": os.environ.get("FABRIC_RELAY_CONFIG_PATH"), - "emitter": "hermes.observability/nemo_relay", + "plugin_config_path": str(self._relay_plugin_config_path), + "emitter": "hermes-agent/nemo-relay", } output["relay_artifacts"] = common_utils.collect_relay_artifacts( self._relay_plugin_config ) return output - def _finalize_relay_session(self) -> None: - if ( - self._relay_plugin_config is None - or self._agent is None - or self._invoke_hook is None - or not self._relay_session_pending - ): - return - if not self._relay_finalize_hook_invoked: - self._invoke_hook( - "on_session_finalize", - session_id=getattr(self._agent, "session_id", ""), - model=getattr(self._agent, "model", None) or self._relay_model_name, - platform=getattr(self._agent, "platform", None) or "fabric", - ) - self._relay_finalize_hook_invoked = True - # Relay subscriber callbacks are queued. The long-lived plugin context - # does not flush them until runtime shutdown, but invocation results - # must include artifacts produced by this turn. - from nemo_relay import subscribers - - subscribers.flush() - self._relay_session_pending = False - self._relay_finalize_hook_invoked = False - async def stop(self) -> None: agent = self._agent session_db = self._session_db - relay_context = self._relay_context - relay_context_entered = self._relay_context_entered - relay_plugin_config = self._relay_plugin_config had_mcp_servers = bool(self._hermes_config.get("mcp_servers")) + previous_relay_environment = self._previous_relay_environment + applied_relay_environment = self._applied_relay_environment errors: list[BaseException] = [] - if relay_plugin_config is not None and agent is not None: - try: - self._finalize_relay_session() - except BaseException as error: - errors.append(error) self._agent = None self._session_db = None self._start_payload = None @@ -496,15 +493,22 @@ async def stop(self) -> None: self._hermes_config = {} self._enabled_toolsets = None self._conversation_history = None - self._relay_context = None - self._relay_context_entered = False - self._relay_session_pending = False - self._relay_finalize_hook_invoked = False - self._invoke_hook = None self._relay_plugin_config = None - self._relay_model_name = "unknown" + self._relay_plugin_config_path = None + self._previous_relay_environment = {} + self._applied_relay_environment = {} self._started = False + for name, applied_value in applied_relay_environment.items(): + current_value = os.environ.get(name) + if current_value != applied_value: + continue + previous_value = previous_relay_environment.get(name) + if previous_value is None: + os.environ.pop(name, None) + else: + os.environ[name] = previous_value + if had_mcp_servers: try: from tools.mcp_tool import shutdown_mcp_servers @@ -524,12 +528,6 @@ async def stop(self) -> None: session_db.close() except BaseException as error: errors.append(error) - if relay_context is not None and relay_context_entered: - try: - await relay_context.__aexit__(None, None, None) - except BaseException as error: - errors.append(error) - if errors: for error in errors: if isinstance(error, asyncio.CancelledError): diff --git a/python/src/nemo_fabric/streaming.py b/python/src/nemo_fabric/streaming.py index 0999d18f..8d9d21ea 100644 --- a/python/src/nemo_fabric/streaming.py +++ b/python/src/nemo_fabric/streaming.py @@ -556,6 +556,9 @@ def _matches_turn_root(self, record: dict[str, Any]) -> bool: self._turn_index is not None and metadata.get("nemo_relay_scope_role") == "turn" and metadata.get("turn_index") == self._turn_index + ) or ( + record.get("name") == "hermes.turn" + and metadata.get("hermes.execution_surface") == "fabric" ) async def close(self) -> None: diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index e32ccacb..a975f6fd 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -10,8 +10,9 @@ import json import os import sys +import tomllib from pathlib import Path -from types import ModuleType, SimpleNamespace +from types import ModuleType from unittest.mock import MagicMock import pytest @@ -58,107 +59,101 @@ def test_validate_hermes_telemetry_provider_rejects_mixed_native_and_relay(): adapter.validate_hermes_telemetry_provider(payload) -def test_finalize_relay_session_flushes_before_artifact_collection(monkeypatch): - calls: list[str] = [] - invoke_hook = MagicMock(side_effect=lambda *args, **kwargs: calls.append("hook")) - runtime = adapter.HermesRuntime() - runtime._relay_plugin_config = {"components": []} - runtime._agent = SimpleNamespace( - session_id="runtime-1", - model="test-model", - platform="fabric", +def test_write_hermes_relay_plugin_config_uses_upstream_toml( + monkeypatch, + tmp_path: Path, +): + relay_config_path = tmp_path / "relay.json" + relay_config_path.write_text( + json.dumps( + { + "relay": { + "config": { + "atof": { + "enabled": True, + "sinks": [{"type": "file"}], + }, + "atif": {"enabled": True}, + } + } + } + ), + encoding="utf-8", ) - runtime._invoke_hook = invoke_hook - runtime._relay_session_pending = True - - from nemo_relay import subscribers - - monkeypatch.setattr(subscribers, "flush", lambda: calls.append("flush")) - - runtime._finalize_relay_session() + monkeypatch.setenv("FABRIC_RELAY_CONFIG_PATH", str(relay_config_path)) + payload = { + "agent_name": "hermes-test-agent", + "base_dir": str(tmp_path), + "config": { + "models": {"default": {"provider": "nvidia", "model": "nvidia/test-model"}} + }, + "runtime_context": {"runtime_id": "runtime-hermes-relay"}, + } - assert calls == ["hook", "flush"] - invoke_hook.assert_called_once_with( - "on_session_finalize", - session_id="runtime-1", - model="test-model", - platform="fabric", + plugin_config_path, plugin_config = adapter.write_hermes_relay_plugin_config( + payload ) - -async def test_stop_does_not_refinalize_completed_relay_turn(monkeypatch): - invoke_hook = MagicMock() - runtime = adapter.HermesRuntime() - runtime._started = True - runtime._relay_plugin_config = {"components": []} - agent = MagicMock( - session_id="runtime-1", - model="test-model", - platform="fabric", + assert plugin_config_path == tmp_path / "relay-config" / "plugins.toml" + with plugin_config_path.open("rb") as stream: + staged_plugin_config = tomllib.load(stream) + assert "atif" not in staged_plugin_config["components"][0]["config"] + assert plugin_config["components"][0]["config"]["atof"]["sinks"][0][ + "output_directory" + ] == str(tmp_path / "artifacts" / "relay" / "runtime-hermes-relay") + relay_environment = adapter.hermes_relay_environment( + plugin_config, + plugin_config_path, ) - session_db = MagicMock() - runtime._agent = agent - runtime._session_db = session_db - runtime._invoke_hook = invoke_hook - runtime._relay_session_pending = True - - from nemo_relay import subscribers - - flush = MagicMock() - monkeypatch.setattr(subscribers, "flush", flush) - - runtime._finalize_relay_session() - await runtime.stop() - - invoke_hook.assert_called_once_with( - "on_session_finalize", - session_id="runtime-1", - model="test-model", - platform="fabric", + assert relay_environment["HERMES_NEMO_RELAY_ATIF_ENABLED"] == "1" + assert relay_environment["HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY"] == str( + tmp_path / "artifacts" / "relay" / "runtime-hermes-relay" ) - flush.assert_called_once_with() - agent.close.assert_called_once_with() - session_db.close.assert_called_once_with() - assert runtime._started is False - assert runtime._relay_session_pending is False - assert runtime._relay_finalize_hook_invoked is False -async def test_stop_retries_failed_relay_flush_without_refinalizing(monkeypatch): - invoke_hook = MagicMock() - runtime = adapter.HermesRuntime() - runtime._started = True - runtime._relay_plugin_config = {"components": []} - runtime._agent = MagicMock( - session_id="runtime-1", - model="test-model", - platform="fabric", - ) - runtime._session_db = MagicMock() - runtime._invoke_hook = invoke_hook - runtime._relay_session_pending = True - - from nemo_relay import subscribers +async def test_runtime_start_stages_upstream_relay_plugin_configuration( + monkeypatch, + tmp_path: Path, +): + plugin_config_path = tmp_path / "relay-config" / "plugins.toml" - flush = MagicMock(side_effect=[RuntimeError("flush failed"), None]) - monkeypatch.setattr(subscribers, "flush", flush) + monkeypatch.setattr( + adapter, + "write_hermes_relay_plugin_config", + lambda _payload: (plugin_config_path, {"version": 1}), + ) - with pytest.raises(RuntimeError, match="flush failed"): - runtime._finalize_relay_session() + def stop_after_staging( + _payload: dict[str, object], + _hermes_home: Path, + *, + relay_enabled: bool, + ) -> tuple[Path, dict[str, object]]: + assert relay_enabled is True + assert os.environ["HERMES_NEMO_RELAY_PLUGINS_TOML"] == str(plugin_config_path) + raise RuntimeError("stop after Relay plugin staging") + + monkeypatch.setattr(adapter, "write_hermes_config", stop_after_staging) + monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "before-start") + payload = { + "base_dir": str(tmp_path), + "config": { + "harness": {"settings": {}}, + "models": {"default": {"provider": "nvidia", "model": "test-model"}}, + }, + "runtime_context": { + "runtime_id": "runtime-relay-plugin", + "environment": {"workspace": str(tmp_path)}, + "artifacts": {"root": str(tmp_path / "artifacts")}, + }, + "telemetry_plan": {"providers": ["relay"], "relay_enabled": True}, + } - assert runtime._relay_session_pending is True - assert runtime._relay_finalize_hook_invoked is True - await runtime.stop() + with pytest.raises(RuntimeError, match="stop after Relay plugin staging"): + await adapter.HermesRuntime().start(payload) - invoke_hook.assert_called_once_with( - "on_session_finalize", - session_id="runtime-1", - model="test-model", - platform="fabric", - ) - assert flush.call_count == 2 - assert runtime._relay_session_pending is False - assert runtime._relay_finalize_hook_invoked is False + assert "HERMES_NEMO_RELAY_PLUGINS_TOML" not in os.environ + assert os.environ["HERMES_NEMO_RELAY_ATIF_ENABLED"] == "before-start" def test_build_hermes_config_maps_fabric_config_to_hermes_config(): @@ -706,9 +701,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history( "base_dir": str(tmp_path), "config": { "harness": {"settings": {}}, - "instructions": { - "system": {"content": "system", "mode": "replace"} - }, + "instructions": {"system": {"content": "system", "mode": "replace"}}, "runtime": {"max_turns": None}, "tools": {"enabled": []}, "models": { @@ -789,6 +782,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history( assert runtime._session_db is None assert runtime._start_payload is None assert runtime._conversation_history is None + assert runtime._relay_plugin_config_path is None assert first["response"] == "first response" assert second["response"] == "second response" assert "session_id" not in second diff --git a/tests/e2e/test_hermes_e2e.py b/tests/e2e/test_hermes_e2e.py index b7c43d1e..963fc8dd 100644 --- a/tests/e2e/test_hermes_e2e.py +++ b/tests/e2e/test_hermes_e2e.py @@ -110,8 +110,13 @@ async def test_mcp_stdio_transport( f"{api_server}/_scenario", json={ "tool_call": { - "name": tool_name, - "arguments": {"timezone": "America/Los_Angeles"}, + # Hermes v0.20 defers MCP schemas behind its native tool-call + # bridge. The bridge dispatches this to the configured MCP tool. + "name": "tool_call", + "arguments": { + "name": tool_name, + "arguments": {"timezone": "America/Los_Angeles"}, + }, } }, timeout=5, @@ -177,12 +182,13 @@ async def test_mcp_stdio_transport( "args": ["-m", "mcp_server_time"], "env": {"MCP_TIME_TEST": "enabled"}, } + assert "mcp_server_time" in result["output"]["enabled_toolsets"] assert {record["scope_category"] for record in tool_records} == {"start", "end"} tool_end = next( record for record in tool_records if record["scope_category"] == "end" ) assert tool_end["category"] == "tool" - assert tool_end["metadata"]["status"] == "ok" + assert tool_end["metadata"]["otel.status_code"] == "OK" assert "America/Los_Angeles" in tool_end["data"] @@ -243,7 +249,7 @@ async def test_artifacts(self): assert output["base_url"] == f"{self.api_server}/v1" assert output["error"] is None assert output["relay_runtime"]["enabled"] is True - assert output["relay_runtime"]["emitter"] == "hermes.observability/nemo_relay" + assert output["relay_runtime"]["emitter"] == "hermes-agent/nemo-relay" assert output["failed"] is False assert "echo user_count=" in output["response"] @@ -257,7 +263,10 @@ async def test_artifacts(self): hermes_config = yaml.safe_load(hermes_config_path.read_text(encoding="utf-8")) assert hermes_config["model"]["provider"] == "nvidia" - assert hermes_config["model"]["default"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" + assert ( + hermes_config["model"]["default"] + == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" + ) assert hermes_config["model"]["base_url"] == f"{self.api_server}/v1" assert hermes_config["plugins"]["enabled"] == ["observability/nemo_relay"] assert output["hermes_native_config"]["plugins"] == ["observability/nemo_relay"] @@ -315,17 +324,41 @@ async def test_atof_artifacts(self): actual_atof_fields = set().union(*(record.keys() for record in atof_records)) assert actual_atof_fields.issuperset(expected_atof_fields) - assert len(atof_records) == 7 + record_kinds = { + (record["name"], record.get("scope_category")) for record in atof_records + } + assert record_kinds.issuperset( + { + ("hermes.session", "start"), + ("hermes.session", "end"), + ("hermes.turn", "start"), + ("hermes.turn", "end"), + ("nvidia", "start"), + ("nvidia", "end"), + } + ) + fabric_scopes = [ + record + for record in atof_records + if record["name"] in {"hermes.session", "hermes.turn"} + ] assert all( - record["metadata"]["model"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" - and record["metadata"]["platform"] == self.atof_platform + record["metadata"]["hermes.execution_surface"] == "fabric" + for record in fabric_scopes + ) + turn_marks = [ + record for record in atof_records + if record["name"] in {"hermes.turn.start", "hermes.turn.end"} + ] + assert all( + record["metadata"]["model"] + == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" + and record["metadata"]["platform"] == self.atof_platform + for record in turn_marks ) - assert atof_records[-2]["name"] == "hermes.session.end" - assert atof_records[-1]["scope_category"] == "end" - async def test_atif_artifacts(self): kinds = {artifact["kind"] for artifact in self.relay_artifacts} assert "atif" in kinds @@ -354,7 +387,12 @@ async def test_atif_artifacts(self): last_step = steps[-1] assert last_step["source"] == "agent" - assert last_step["message"] == self.output["response"] - assert last_step["model_name"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" + # The upstream exporter derives this field from the provider's final + # wire response. The mock streaming response has no final text field, + # while Fabric's normalized response is assembled from its deltas. + assert last_step["message"] in {"", self.output["response"]} + assert ( + last_step["model_name"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" + ) assert last_step["extra"]["invocation"]["framework"] == "nemo_relay" assert last_step["extra"]["invocation"]["status"] == "completed" diff --git a/tests/e2e/test_hermes_runtime.py b/tests/e2e/test_hermes_runtime.py index 4f0e57a0..791e88ee 100644 --- a/tests/e2e/test_hermes_runtime.py +++ b/tests/e2e/test_hermes_runtime.py @@ -31,8 +31,6 @@ async def test_hermes_runtime(): async def test_hermes_runtime_with_relay(): _require_hermes_integration() - if importlib.util.find_spec("nemo_relay") is None: - pytest.fail("the nemo-relay Python package is required") await _run(relay=True) @@ -103,8 +101,7 @@ async def _run(*, relay: bool) -> None: for result in (r1, r2): assert result.telemetry[0].provider == "relay", result.to_mapping() assert { - artifact["kind"] - for artifact in result["output"]["relay_artifacts"] + artifact["kind"] for artifact in result["output"]["relay_artifacts"] } >= {"atof", "atif"}, result.to_mapping() assert runtime.status is RuntimeStatus.STOPPED, runtime.status diff --git a/tests/python/test_streaming.py b/tests/python/test_streaming.py index 31736cec..64a7fe1d 100644 --- a/tests/python/test_streaming.py +++ b/tests/python/test_streaming.py @@ -880,14 +880,16 @@ async def produce() -> None: @pytest.mark.parametrize( - "current_metadata", + ("current_metadata", "current_name"), [ - {"nemo_fabric_request_id": "request-2"}, - {"nemo_relay_scope_role": "turn", "turn_index": 2}, + ({"nemo_fabric_request_id": "request-2"}, None), + ({"nemo_relay_scope_role": "turn", "turn_index": 2}, None), + ({"hermes.execution_surface": "fabric"}, "hermes.turn"), ], ) async def test_listener_correlates_records_to_active_turn( current_metadata: dict[str, Any], + current_name: str | None, ): listener = await _AtofStreamListener(maxsize=4).start() listener.begin_stream(request_id="request-2", turn_index=2) @@ -896,6 +898,7 @@ async def test_listener_correlates_records_to_active_turn( "kind": "scope", "scope_category": "start", "uuid": "current", + "name": current_name, "metadata": current_metadata, }, { From 1c911d1bc444ada31159997022a1bd45860521ed Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 12:42:40 -0700 Subject: [PATCH 02/12] fix(hermes): stage Relay v3 plugin configuration Signed-off-by: Zhongxuan Wang --- .../src/nemo_fabric_adapters/common/utils.py | 4 +- adapters/hermes/pyproject.toml | 4 +- .../nemo_fabric_adapters/hermes/adapter.py | 97 +++++++++++-------- adapters/hermes/uv.lock | 22 ++--- .../adapters/test_adapter_package_metadata.py | 6 +- tests/adapters/test_hermes_adapter.py | 71 ++++++++++++-- uv.lock | 4 +- 7 files changed, 139 insertions(+), 69 deletions(-) diff --git a/adapters/common/src/nemo_fabric_adapters/common/utils.py b/adapters/common/src/nemo_fabric_adapters/common/utils.py index 96ae9c72..e44d1cc6 100644 --- a/adapters/common/src/nemo_fabric_adapters/common/utils.py +++ b/adapters/common/src/nemo_fabric_adapters/common/utils.py @@ -219,9 +219,11 @@ def merge_unique(*values: Any) -> list[str]: merged.append(item) return merged + def without_none(mapping: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in mapping.items() if value is not None} + def dump_yaml(value: dict[str, Any]) -> str: try: import yaml @@ -409,7 +411,7 @@ def write_relay_configs( relay_config_path.write_text(tomli_w.dumps(relay_config), encoding="utf-8") if plugin_config is not None: - if observability_version != 2: + if observability_version not in {2, 3}: raise ValueError( f"unsupported NeMo Relay observability config version {observability_version}" ) diff --git a/adapters/hermes/pyproject.toml b/adapters/hermes/pyproject.toml index a308eded..11fad367 100644 --- a/adapters/hermes/pyproject.toml +++ b/adapters/hermes/pyproject.toml @@ -34,11 +34,11 @@ harness = [ "hermes-agent[mcp]>=0.19.0; python_version < '3.14'", ] relay = [ - "nemo-relay>=0.6.0,<0.7", + "nemo-relay==0.7.2", ] full = [ "hermes-agent[mcp]>=0.19.0; python_version < '3.14'", - "nemo-relay>=0.6.0,<0.7", + "nemo-relay==0.7.2", ] [project.urls] diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index 648269cc..2f1b9af5 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -154,50 +154,50 @@ def write_hermes_relay_plugin_config( if component.get("kind") != "observability": continue observability = component.get("config") - if isinstance(observability, dict): - # Hermes v0.20.0 manages ATIF through its plugin environment. Keep - # the complete Fabric config for artifact collection, but do not - # configure ATIF twice through the component TOML. - observability.pop("atif", None) + if not isinstance(observability, dict): + continue + + if observability.get("version") != 3: + # Relay 0.7 combines Fabric's legacy OTLP and OpenInference exporter + # settings into typed OpenTelemetry endpoints in its v3 schema. + endpoints = [] + for config_name, endpoint_type in ( + ("opentelemetry", "full"), + ("openinference", "openinference"), + ): + exporter = observability.pop(config_name, None) + if not isinstance(exporter, dict) or not exporter.get("enabled"): + continue + endpoint = { + key: value + for key, value in exporter.items() + if key != "enabled" and value is not None + } + endpoint["type"] = endpoint_type + endpoints.append(endpoint) + if endpoints: + observability["opentelemetry"] = { + "enabled": True, + "endpoints": endpoints, + } + observability["version"] = 3 + + # Fabric finalizes Hermes' Relay session after every invocation. Each + # finalization reinitializes Relay for the next turn, so a file sink + # cannot overwrite the runtime-scoped artifact it created previously. + for sink in (observability.get("atof") or {}).get("sinks") or []: + if isinstance(sink, dict) and sink.get("type") == "file": + if sink.get("mode") == "overwrite": + sink["mode"] = "append" _, plugin_config_path = common_utils.write_relay_configs( - plugin_config=hermes_plugin_config + plugin_config=hermes_plugin_config, + observability_version=3, ) if plugin_config_path is None: raise RuntimeError("Hermes Relay plugin configuration was not generated") return plugin_config_path, plugin_config -def hermes_relay_environment( - plugin_config: dict[str, Any], - plugin_config_path: Path, -) -> dict[str, str]: - """Translate Fabric's ATIF settings to Hermes' upstream plugin environment.""" - - environment = {"HERMES_NEMO_RELAY_PLUGINS_TOML": str(plugin_config_path)} - for component in plugin_config.get("components", []): - if component.get("kind") != "observability": - continue - observability = component.get("config") - if not isinstance(observability, dict): - continue - atif = observability.get("atif") - if not isinstance(atif, dict) or not atif.get("enabled"): - continue - environment["HERMES_NEMO_RELAY_ATIF_ENABLED"] = "1" - for config_key, environment_key in ( - ("output_directory", "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY"), - ("filename_template", "HERMES_NEMO_RELAY_ATIF_FILENAME_TEMPLATE"), - ("agent_name", "HERMES_NEMO_RELAY_ATIF_AGENT_NAME"), - ("agent_version", "HERMES_NEMO_RELAY_ATIF_AGENT_VERSION"), - ("model_name", "HERMES_NEMO_RELAY_ATIF_MODEL_NAME"), - ): - value = atif.get(config_key) - if value is not None: - environment[environment_key] = str(value) - break - return environment - - def hermes_mcp_server_config(server: dict[str, Any]) -> dict[str, Any]: transport = str(server.get("transport") or "").strip().lower() raw_target = server.get("url") @@ -314,10 +314,11 @@ async def start(self, payload: dict[str, Any]) -> None: self._relay_plugin_config_path, self._relay_plugin_config, ) = write_hermes_relay_plugin_config(payload) - relay_environment = hermes_relay_environment( - self._relay_plugin_config, - self._relay_plugin_config_path, - ) + relay_environment = { + "HERMES_NEMO_RELAY_PLUGINS_TOML": str( + self._relay_plugin_config_path + ) + } self._previous_relay_environment = { name: os.environ.get(name) for name in HERMES_RELAY_ENV_NAMES } @@ -428,12 +429,24 @@ async def invoke(self, invocation: dict[str, Any]) -> dict[str, Any]: user_message = json.dumps(user_message, sort_keys=True) def invoke_turn() -> tuple[dict[str, Any], str]: - return _invoke_hermes_turn( + result, adapter_stdout = _invoke_hermes_turn( agent=self._agent, system_prompt=common_utils.system_instruction(start_payload), user_message=user_message, conversation_history=self._conversation_history, ) + if self._relay_plugin_config is not None: + # Relay 0.7 writes TOML-configured ATIF at the upstream Hermes + # session-finalization boundary. Fabric defines each invoke as + # an artifact-complete boundary, so finalize through Hermes' + # lifecycle instead of reaching into Relay directly. + from hermes_cli.lifecycle import finalize_session + + finalize_session( + session_id=str(self._agent.session_id), + platform="fabric", + ) + return result, adapter_stdout # Hermes' upstream Relay integration drives async Relay hooks from its # synchronous agent loop. Run that loop outside this lifecycle server's diff --git a/adapters/hermes/uv.lock b/adapters/hermes/uv.lock index 2a501733..6fecf041 100644 --- a/adapters/hermes/uv.lock +++ b/adapters/hermes/uv.lock @@ -647,24 +647,24 @@ requires-dist = [ { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'full'", specifier = ">=0.19.0" }, { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'harness'", specifier = ">=0.19.0" }, { name = "nemo-fabric-adapters-common", editable = "../common" }, - { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.7" }, - { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.7" }, + { name = "nemo-relay", marker = "extra == 'full'", specifier = "==0.7.2" }, + { name = "nemo-relay", marker = "extra == 'relay'", specifier = "==0.7.2" }, ] provides-extras = ["harness", "relay", "full"] [[package]] name = "nemo-relay" -version = "0.6.0" +version = "0.7.2" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/16/db/44d7258ee620c5cce6dc588983fcb11be3ee03ae71089a65686c14d9ee02/nemo_relay-0.6.0.tar.gz", hash = "sha256:f3d3088019609bc953357b5598a47481dc3e7dc8f11ecf27002ede251f37eb7b", size = 1071046, upload-time = "2026-08-03T14:55:49.702Z" } +sdist = { url = "https://files.pythonhosted.org/packages/58/81/a7a545ac3a2f8c670d261c89df599aa8fbf49d8be45fd1f52efb36b489eb/nemo_relay-0.7.2.tar.gz", hash = "sha256:828d9f6c7d7e4e42276bb7192bd44202c761e0c76fa4943d84e051b5a99028e5", size = 1295616, upload-time = "2026-08-08T01:54:00.953Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/25/65/d320016505457cc30971f575e8dadffb923b7cfc780ab8bb25a4ce9d305c/nemo_relay-0.6.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:ad5dae6febf6532d7b113abc2a404679c8feffc499df3034b93d9a078185d2bb", size = 9917779, upload-time = "2026-07-22T20:07:48.961Z" }, - { url = "https://files.pythonhosted.org/packages/ae/c0/f33250e71c4206da1b339072893f9a1e39295fe1aceb9a2fef4b8620a0f2/nemo_relay-0.6.0-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c0cd9570f64c6956fe3bfb82af1cdb3ee70cb50b51098cdb0de831c3f9b4e904", size = 8888375, upload-time = "2026-07-22T20:07:51.049Z" }, - { url = "https://files.pythonhosted.org/packages/a3/f4/d1dfaed022da0f6f14765a122867f976a69cc520fe1faaf99757f5719d1f/nemo_relay-0.6.0-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:849daa9e45158ac581e54506e0fcc7a24f557d1ed06dbdc074f5de7a00393cbc", size = 9336372, upload-time = "2026-07-22T20:07:53.224Z" }, - { url = "https://files.pythonhosted.org/packages/60/9e/f8b80509eef5e05702b940a3d1e2f60c962548d87dec2d712fd3804e6cd4/nemo_relay-0.6.0-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:8c80e534b76bb0455cfc222aaf5c10fa064c088b3c0d5e137eb3c97db46dbc47", size = 10578834, upload-time = "2026-07-30T15:44:44.726Z" }, - { url = "https://files.pythonhosted.org/packages/f0/a0/84ee49d45a1a874457f2f9260d30fd573af30c9be360d9d97b8bb7835ad9/nemo_relay-0.6.0-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:c8bc4a792a2f8c35ddef1b900cc43be2b2bbbfcd5e7cf65aa0829c04d25eb77f", size = 10849651, upload-time = "2026-07-30T15:44:40.48Z" }, - { url = "https://files.pythonhosted.org/packages/3a/b0/908d77f75b9054e1e403da78f7d9430249a45939070d4298abbb41a04b5b/nemo_relay-0.6.0-cp311-abi3-win_amd64.whl", hash = "sha256:bfbbedfd130fa95c9b8c04643c30910df850e0ed3500beeab82b00ca2d94e7ea", size = 9613425, upload-time = "2026-07-22T20:07:55.175Z" }, - { url = "https://files.pythonhosted.org/packages/cd/71/c438b9d746303ff7f270d99f13b250bf947cdac3e52de2a83fd132cbca0b/nemo_relay-0.6.0-cp311-abi3-win_arm64.whl", hash = "sha256:82fe132943399d89e6ec34dc28df0be7bbe41b84f6698c545928b8816b6010f6", size = 9034810, upload-time = "2026-07-22T20:07:57.467Z" }, + { url = "https://files.pythonhosted.org/packages/5a/cd/f50440257f01bc5ab3d668331c90e06cf4edcc84dc7dc582d322ad05b622/nemo_relay-0.7.2-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:e7c7977f0903793cc34c5542bf2b2e44d107def8a5ae9f1b28f06dd61ddec4ed", size = 9246341, upload-time = "2026-08-08T01:53:19.832Z" }, + { url = "https://files.pythonhosted.org/packages/ed/9f/4041446dd134218799a34b5b5fad3a62d3e1d0a6c322ba2ca4b896ba1393/nemo_relay-0.7.2-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b4ae77c1f3d58eabda264e82ffaca54548df80caede7dd6af8cbd8f72b4a82ed", size = 8454070, upload-time = "2026-08-08T01:53:22.524Z" }, + { url = "https://files.pythonhosted.org/packages/11/83/90230c2e9fae1aee39f768d4a9ef57e9f2716bcaed1a5923cce8b526c66b/nemo_relay-0.7.2-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:0ce7103aec546766649c182619d16aa6ad07439e4d0ebd16d95c5004afb3e56a", size = 8954377, upload-time = "2026-08-08T01:53:25.267Z" }, + { url = "https://files.pythonhosted.org/packages/71/e7/463fa461d0801146fec6a00cbc02e8961b30089d65ba170f9dfa9e6e3dcd/nemo_relay-0.7.2-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:2e7d0c2629ade7313aaed71d2272dca96a2fafad0248d0f87cf40a7720b252a0", size = 10322132, upload-time = "2026-08-08T01:53:27.991Z" }, + { url = "https://files.pythonhosted.org/packages/32/8c/e20ec9c52bd1edd953157aaf24d0d9f9ab8afcbf108fc2356f398e252da8/nemo_relay-0.7.2-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:b841c92395d7686c7f233036008294b9d362af1ec5123ab0babbfd11cbb04054", size = 10704141, upload-time = "2026-08-08T01:53:30.453Z" }, + { url = "https://files.pythonhosted.org/packages/5a/c1/92a73961ea759b433b1f897b225662d499123cb962b48dc8ece19f610a09/nemo_relay-0.7.2-cp311-abi3-win_amd64.whl", hash = "sha256:0cdcc5e09d6d62d5c1d385dc62c9233eb714a25f36a09da81e5b9731e3c67903", size = 8803938, upload-time = "2026-08-08T01:53:33.437Z" }, + { url = "https://files.pythonhosted.org/packages/9d/ec/2de114dab437431173988b9b11f46e8d377e12d57e1b4903258f3e03c2df/nemo_relay-0.7.2-cp311-abi3-win_arm64.whl", hash = "sha256:ca5f66e617311f836a10d96f120f3f32a99b4267d65048453b31951de3419a9d", size = 8438997, upload-time = "2026-08-08T01:53:36.12Z" }, ] [[package]] diff --git a/tests/adapters/test_adapter_package_metadata.py b/tests/adapters/test_adapter_package_metadata.py index 43e46089..af5906a3 100644 --- a/tests/adapters/test_adapter_package_metadata.py +++ b/tests/adapters/test_adapter_package_metadata.py @@ -50,10 +50,8 @@ def load_pyproject(path: str) -> dict: f"nemo-fabric-adapters-hermes[harness] == {PACKAGE_VERSION}; " "python_version < '3.14'" ), - "harness": [ - "hermes-agent[mcp]>=0.19.0; python_version < '3.14'" - ], - "relay": ["nemo-relay>=0.6.0,<0.7"], + "harness": ["hermes-agent[mcp]>=0.19.0; python_version < '3.14'"], + "relay": ["nemo-relay==0.7.2"], }, } diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index a975f6fd..fde1a157 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -97,18 +97,75 @@ def test_write_hermes_relay_plugin_config_uses_upstream_toml( assert plugin_config_path == tmp_path / "relay-config" / "plugins.toml" with plugin_config_path.open("rb") as stream: staged_plugin_config = tomllib.load(stream) - assert "atif" not in staged_plugin_config["components"][0]["config"] + staged_observability = staged_plugin_config["components"][0]["config"] + assert staged_observability["version"] == 3 + assert staged_observability["atif"]["enabled"] is True + assert staged_observability["atof"]["sinks"][0]["mode"] == "append" assert plugin_config["components"][0]["config"]["atof"]["sinks"][0][ "output_directory" ] == str(tmp_path / "artifacts" / "relay" / "runtime-hermes-relay") - relay_environment = adapter.hermes_relay_environment( - plugin_config, - plugin_config_path, + assert ( + plugin_config["components"][0]["config"]["atof"]["sinks"][0]["mode"] + == "overwrite" ) - assert relay_environment["HERMES_NEMO_RELAY_ATIF_ENABLED"] == "1" - assert relay_environment["HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY"] == str( - tmp_path / "artifacts" / "relay" / "runtime-hermes-relay" + + +def test_write_hermes_relay_plugin_config_migrates_otlp_exporters_to_relay_v3( + monkeypatch, + tmp_path: Path, +): + relay_config_path = tmp_path / "relay.json" + relay_config_path.write_text( + json.dumps( + { + "relay": { + "config": { + "opentelemetry": { + "enabled": True, + "endpoint": "https://otel.example/v1/traces", + "service_name": "fabric", + }, + "openinference": { + "enabled": True, + "endpoint": "https://openinference.example/v1/traces", + "service_name": "fabric", + }, + } + } + } + ), + encoding="utf-8", ) + monkeypatch.setenv("FABRIC_RELAY_CONFIG_PATH", str(relay_config_path)) + payload = { + "agent_name": "hermes-test-agent", + "base_dir": str(tmp_path), + "config": { + "models": {"default": {"provider": "nvidia", "model": "nvidia/test-model"}} + }, + "runtime_context": {"runtime_id": "runtime-hermes-relay"}, + } + + plugin_config_path, _ = adapter.write_hermes_relay_plugin_config(payload) + + with plugin_config_path.open("rb") as stream: + staged_observability = tomllib.load(stream)["components"][0]["config"] + assert staged_observability["version"] == 3 + assert staged_observability["opentelemetry"] == { + "enabled": True, + "endpoints": [ + { + "type": "full", + "endpoint": "https://otel.example/v1/traces", + "service_name": "fabric", + }, + { + "type": "openinference", + "endpoint": "https://openinference.example/v1/traces", + "service_name": "fabric", + }, + ], + } async def test_runtime_start_stages_upstream_relay_plugin_configuration( diff --git a/uv.lock b/uv.lock index 3def94c4..639766ce 100644 --- a/uv.lock +++ b/uv.lock @@ -2359,8 +2359,8 @@ requires-dist = [ { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'full'", specifier = ">=0.19.0" }, { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'harness'", specifier = ">=0.19.0" }, { name = "nemo-fabric-adapters-common", editable = "adapters/common" }, - { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.7" }, - { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.7" }, + { name = "nemo-relay", marker = "extra == 'full'", specifier = "==0.7.2" }, + { name = "nemo-relay", marker = "extra == 'relay'", specifier = "==0.7.2" }, ] provides-extras = ["harness", "relay", "full"] From 072ab716b17030976f8ce21aa3a6c08777a90317 Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 12:59:52 -0700 Subject: [PATCH 03/12] fix(hermes): support Relay compatibility range Signed-off-by: Zhongxuan Wang --- adapters/hermes/pyproject.toml | 4 ++-- .../src/nemo_fabric_adapters/hermes/adapter.py | 15 +++++++++++++-- adapters/hermes/uv.lock | 4 ++-- tests/adapters/test_adapter_package_metadata.py | 2 +- tests/adapters/test_hermes_adapter.py | 14 +++++++++++++- uv.lock | 4 ++-- 6 files changed, 33 insertions(+), 10 deletions(-) diff --git a/adapters/hermes/pyproject.toml b/adapters/hermes/pyproject.toml index 11fad367..5d2f215e 100644 --- a/adapters/hermes/pyproject.toml +++ b/adapters/hermes/pyproject.toml @@ -34,11 +34,11 @@ harness = [ "hermes-agent[mcp]>=0.19.0; python_version < '3.14'", ] relay = [ - "nemo-relay==0.7.2", + "nemo-relay>=0.6.0,<0.8", ] full = [ "hermes-agent[mcp]>=0.19.0; python_version < '3.14'", - "nemo-relay==0.7.2", + "nemo-relay>=0.6.0,<0.8", ] [project.urls] diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index 2f1b9af5..dcc56626 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -17,6 +17,7 @@ import logging import os from contextlib import redirect_stdout +from importlib.metadata import version as distribution_version from io import StringIO from pathlib import Path from typing import Any @@ -150,6 +151,16 @@ def write_hermes_relay_plugin_config( plugin_config = common_utils.load_relay_plugin_config(payload) hermes_plugin_config = copy.deepcopy(plugin_config) + relay_version = distribution_version("nemo-relay") + try: + relay_major, relay_minor = ( + int(part) for part in relay_version.split(".", maxsplit=2)[:2] + ) + except ValueError as error: + raise RuntimeError( + f"unsupported NeMo Relay version {relay_version!r}" + ) from error + observability_version = 3 if (relay_major, relay_minor) >= (0, 7) else 2 for component in hermes_plugin_config.get("components", []): if component.get("kind") != "observability": continue @@ -157,7 +168,7 @@ def write_hermes_relay_plugin_config( if not isinstance(observability, dict): continue - if observability.get("version") != 3: + if observability_version == 3 and observability.get("version") != 3: # Relay 0.7 combines Fabric's legacy OTLP and OpenInference exporter # settings into typed OpenTelemetry endpoints in its v3 schema. endpoints = [] @@ -191,7 +202,7 @@ def write_hermes_relay_plugin_config( sink["mode"] = "append" _, plugin_config_path = common_utils.write_relay_configs( plugin_config=hermes_plugin_config, - observability_version=3, + observability_version=observability_version, ) if plugin_config_path is None: raise RuntimeError("Hermes Relay plugin configuration was not generated") diff --git a/adapters/hermes/uv.lock b/adapters/hermes/uv.lock index 6fecf041..383582c0 100644 --- a/adapters/hermes/uv.lock +++ b/adapters/hermes/uv.lock @@ -647,8 +647,8 @@ requires-dist = [ { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'full'", specifier = ">=0.19.0" }, { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'harness'", specifier = ">=0.19.0" }, { name = "nemo-fabric-adapters-common", editable = "../common" }, - { name = "nemo-relay", marker = "extra == 'full'", specifier = "==0.7.2" }, - { name = "nemo-relay", marker = "extra == 'relay'", specifier = "==0.7.2" }, + { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.8" }, + { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.8" }, ] provides-extras = ["harness", "relay", "full"] diff --git a/tests/adapters/test_adapter_package_metadata.py b/tests/adapters/test_adapter_package_metadata.py index af5906a3..e7a7f44d 100644 --- a/tests/adapters/test_adapter_package_metadata.py +++ b/tests/adapters/test_adapter_package_metadata.py @@ -51,7 +51,7 @@ def load_pyproject(path: str) -> dict: "python_version < '3.14'" ), "harness": ["hermes-agent[mcp]>=0.19.0; python_version < '3.14'"], - "relay": ["nemo-relay==0.7.2"], + "relay": ["nemo-relay>=0.6.0,<0.8"], }, } diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index fde1a157..7b18f92d 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -63,6 +63,7 @@ def test_write_hermes_relay_plugin_config_uses_upstream_toml( monkeypatch, tmp_path: Path, ): + monkeypatch.setattr(adapter, "distribution_version", lambda _name: "0.6.0") relay_config_path = tmp_path / "relay.json" relay_config_path.write_text( json.dumps( @@ -74,6 +75,11 @@ def test_write_hermes_relay_plugin_config_uses_upstream_toml( "sinks": [{"type": "file"}], }, "atif": {"enabled": True}, + "opentelemetry": { + "enabled": True, + "endpoint": "https://otel.example/v1/traces", + "service_name": "fabric", + }, } } } @@ -98,9 +104,14 @@ def test_write_hermes_relay_plugin_config_uses_upstream_toml( with plugin_config_path.open("rb") as stream: staged_plugin_config = tomllib.load(stream) staged_observability = staged_plugin_config["components"][0]["config"] - assert staged_observability["version"] == 3 + assert staged_observability["version"] == 2 assert staged_observability["atif"]["enabled"] is True assert staged_observability["atof"]["sinks"][0]["mode"] == "append" + assert staged_observability["opentelemetry"] == { + "enabled": True, + "endpoint": "https://otel.example/v1/traces", + "service_name": "fabric", + } assert plugin_config["components"][0]["config"]["atof"]["sinks"][0][ "output_directory" ] == str(tmp_path / "artifacts" / "relay" / "runtime-hermes-relay") @@ -114,6 +125,7 @@ def test_write_hermes_relay_plugin_config_migrates_otlp_exporters_to_relay_v3( monkeypatch, tmp_path: Path, ): + monkeypatch.setattr(adapter, "distribution_version", lambda _name: "0.7.2") relay_config_path = tmp_path / "relay.json" relay_config_path.write_text( json.dumps( diff --git a/uv.lock b/uv.lock index 639766ce..6f530c90 100644 --- a/uv.lock +++ b/uv.lock @@ -2359,8 +2359,8 @@ requires-dist = [ { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'full'", specifier = ">=0.19.0" }, { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'harness'", specifier = ">=0.19.0" }, { name = "nemo-fabric-adapters-common", editable = "adapters/common" }, - { name = "nemo-relay", marker = "extra == 'full'", specifier = "==0.7.2" }, - { name = "nemo-relay", marker = "extra == 'relay'", specifier = "==0.7.2" }, + { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.8" }, + { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.8" }, ] provides-extras = ["harness", "relay", "full"] From 629d1c126fed314fd5841015c3997f8b7616fa0e Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 13:11:55 -0700 Subject: [PATCH 04/12] fix(hermes): clarify Relay session finalization Signed-off-by: Zhongxuan Wang --- .../hermes/src/nemo_fabric_adapters/hermes/adapter.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index dcc56626..85bdccb3 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -447,10 +447,11 @@ def invoke_turn() -> tuple[dict[str, Any], str]: conversation_history=self._conversation_history, ) if self._relay_plugin_config is not None: - # Relay 0.7 writes TOML-configured ATIF at the upstream Hermes - # session-finalization boundary. Fabric defines each invoke as - # an artifact-complete boundary, so finalize through Hermes' - # lifecycle instead of reaching into Relay directly. + # Hermes writes TOML-configured ATIF at its session-finalization + # boundary for every supported Relay version. Fabric defines + # each invoke as an artifact-complete boundary, so finalize + # through Hermes' lifecycle instead of reaching into Relay + # directly. from hermes_cli.lifecycle import finalize_session finalize_session( From efa0ef47f71e87e313ccc88a0ccee3b5916b53e5 Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 13:55:47 -0700 Subject: [PATCH 05/12] fix(hermes): support published Relay integration Signed-off-by: Zhongxuan Wang --- .../nemo_fabric_adapters/hermes/adapter.py | 111 ++++++++++++++---- python/src/nemo_fabric/streaming.py | 25 +++- tests/adapters/test_hermes_adapter.py | 89 +++++++++++++- tests/adapters/test_hermes_streaming.py | 6 +- tests/e2e/test_hermes_e2e.py | 72 ++++++------ tests/python/test_streaming.py | 44 +++++++ 6 files changed, 288 insertions(+), 59 deletions(-) diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index 85bdccb3..9ccd1e62 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -37,6 +37,9 @@ "openai": "OPENAI_API_KEY", "openrouter": "OPENROUTER_API_KEY", } +# Hermes still discovers its Relay plugin through the TOML-path environment +# variable. Keep the related fallback names here solely to clear and restore +# inherited process state; Fabric does not translate its Relay config into them. HERMES_RELAY_ENV_NAMES = ( "HERMES_NEMO_RELAY_PLUGINS_TOML", "HERMES_NEMO_RELAY_ATIF_ENABLED", @@ -48,6 +51,41 @@ ) +def finalize_hermes_relay_session(session_id: str) -> None: + """Finalize one Relay session through the installed Hermes lifecycle API.""" + try: + from hermes_cli.lifecycle import finalize_session + except ModuleNotFoundError as error: + if error.name != "hermes_cli.lifecycle": + raise + # Hermes 0.19 exposes the same finalization boundary as a plugin hook. + from hermes_cli.plugins import invoke_hook + + invoke_hook("on_session_finalize", session_id=session_id, platform="fabric") + else: + finalize_session(session_id=session_id, platform="fabric") + + +def _fabric_stream_sink_enabled(config: dict[str, Any] | None) -> bool: + if config is None: + return False + for component in config.get("components") or []: + if not isinstance(component, dict) or component.get("kind") != "observability": + continue + component_config = component.get("config") + if not isinstance(component_config, dict): + continue + atof = component_config.get("atof") + if not isinstance(atof, dict): + continue + if any( + isinstance(sink, dict) and sink.get("name") == "nemo-fabric-stream" + for sink in atof.get("sinks") or [] + ): + return True + return False + + def _api_key_env(model_config: dict[str, Any]) -> str: explicit = model_config.get("api_key_env") if isinstance(explicit, str) and explicit: @@ -291,6 +329,7 @@ def __init__(self) -> None: self._relay_plugin_config_path: Path | None = None self._previous_relay_environment: dict[str, str | None] = {} self._applied_relay_environment: dict[str, str | None] = {} + self._active_invoke_task: asyncio.Task[tuple[dict[str, Any], str]] | None = None async def start(self, payload: dict[str, Any]) -> None: if self._started: @@ -439,31 +478,51 @@ async def invoke(self, invocation: dict[str, Any]) -> dict[str, Any]: if not isinstance(user_message, str): user_message = json.dumps(user_message, sort_keys=True) - def invoke_turn() -> tuple[dict[str, Any], str]: - result, adapter_stdout = _invoke_hermes_turn( - agent=self._agent, - system_prompt=common_utils.system_instruction(start_payload), - user_message=user_message, - conversation_history=self._conversation_history, - ) - if self._relay_plugin_config is not None: - # Hermes writes TOML-configured ATIF at its session-finalization - # boundary for every supported Relay version. Fabric defines - # each invoke as an artifact-complete boundary, so finalize - # through Hermes' lifecycle instead of reaching into Relay - # directly. - from hermes_cli.lifecycle import finalize_session - - finalize_session( - session_id=str(self._agent.session_id), - platform="fabric", + def run_hermes_turn() -> tuple[dict[str, Any], str]: + try: + return _invoke_hermes_turn( + agent=self._agent, + system_prompt=common_utils.system_instruction(start_payload), + user_message=user_message, + conversation_history=self._conversation_history, ) - return result, adapter_stdout + finally: + if self._relay_plugin_config is not None: + # Hermes writes TOML-configured ATIF at its session-finalization + # boundary for every supported Relay version. Fabric defines + # each invoke as an artifact-complete boundary, so finalize + # through Hermes' lifecycle instead of reaching into Relay + # directly. + finalize_hermes_relay_session(str(self._agent.session_id)) + + def invoke_turn() -> tuple[dict[str, Any], str]: + if not _fabric_stream_sink_enabled(self._relay_plugin_config): + return run_hermes_turn() + + from nemo_relay import ScopeType, scope + + with scope.scope( + "nemo-fabric-invocation", + ScopeType.Agent, + metadata={"nemo_fabric_request_id": request.get("request_id")}, + ): + return run_hermes_turn() # Hermes' upstream Relay integration drives async Relay hooks from its # synchronous agent loop. Run that loop outside this lifecycle server's # event-loop thread so Hermes can own its Relay event loop. - result, adapter_stdout = await asyncio.to_thread(invoke_turn) + if self._active_invoke_task is not None: + raise lifecycle.LifecycleError( + "hermes_invocation_in_progress", + "Hermes runtime already has an active invocation", + ) + invoke_task = asyncio.create_task(asyncio.to_thread(invoke_turn)) + self._active_invoke_task = invoke_task + try: + result, adapter_stdout = await asyncio.shield(invoke_task) + finally: + if invoke_task.done() and self._active_invoke_task is invoke_task: + self._active_invoke_task = None messages = result.get("messages") or [] if isinstance(messages, list): self._conversation_history = messages @@ -500,12 +559,22 @@ def invoke_turn() -> tuple[dict[str, Any], str]: return output async def stop(self) -> None: + active_invoke_task = self._active_invoke_task + errors: list[BaseException] = [] + if active_invoke_task is not None: + try: + await asyncio.shield(active_invoke_task) + except BaseException as error: + errors.append(error) + finally: + if self._active_invoke_task is active_invoke_task: + self._active_invoke_task = None + agent = self._agent session_db = self._session_db had_mcp_servers = bool(self._hermes_config.get("mcp_servers")) previous_relay_environment = self._previous_relay_environment applied_relay_environment = self._applied_relay_environment - errors: list[BaseException] = [] self._agent = None self._session_db = None self._start_payload = None diff --git a/python/src/nemo_fabric/streaming.py b/python/src/nemo_fabric/streaming.py index 8d9d21ea..8897d737 100644 --- a/python/src/nemo_fabric/streaming.py +++ b/python/src/nemo_fabric/streaming.py @@ -255,6 +255,7 @@ def __init__( self._request_id: str | None = None self._turn_index: int | None = None self._turn_root_uuid: str | None = None + self._legacy_hermes_turn_id: str | None = None self._turn_scope_uuids: set[str] = set() self._saw_atof_data = False self._matched_turn_root = False @@ -308,6 +309,7 @@ def begin_stream( self._request_id = request_id self._turn_index = turn_index self._turn_root_uuid = None + self._legacy_hermes_turn_id = None self._turn_scope_uuids.clear() self._saw_atof_data = False self._matched_turn_root = request_id is None and turn_index is None @@ -322,6 +324,7 @@ def end_stream(self) -> None: self._request_id = None self._turn_index = None self._turn_root_uuid = None + self._legacy_hermes_turn_id = None self._turn_scope_uuids.clear() def warn_if_unavailable(self) -> None: @@ -521,10 +524,21 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool: uuid = record.get("uuid") if not isinstance(uuid, str): return False + metadata = record.get("metadata") + if ( + self._legacy_hermes_turn_id is not None + and isinstance(metadata, dict) + and metadata.get("turn_id") == self._legacy_hermes_turn_id + ): + return True if self._turn_root_uuid is None: if not self._matches_turn_root(record): return False self._turn_root_uuid = uuid + if isinstance(metadata, dict) and record.get("kind") == "mark": + turn_id = metadata.get("turn_id") + if isinstance(turn_id, str): + self._legacy_hermes_turn_id = turn_id self._turn_scope_uuids.add(uuid) self._matched_turn_root = True return True @@ -542,11 +556,18 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool: return True def _matches_turn_root(self, record: dict[str, Any]) -> bool: - if record.get("kind") != "scope" or record.get("scope_category") != "start": - return False metadata = record.get("metadata") if not isinstance(metadata, dict): return False + if ( + record.get("kind") == "mark" + and record.get("name") == "hermes.turn.start" + and metadata.get("platform") == "fabric" + and isinstance(metadata.get("turn_id"), str) + ): + return True + if record.get("kind") != "scope" or record.get("scope_category") != "start": + return False if ( self._request_id is not None and metadata.get("nemo_fabric_request_id") == self._request_id diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index 7b18f92d..a0d1baab 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -5,11 +5,13 @@ from __future__ import annotations +import asyncio import importlib.util import inspect import json import os import sys +import threading import tomllib from pathlib import Path from types import ModuleType @@ -180,6 +182,23 @@ def test_write_hermes_relay_plugin_config_migrates_otlp_exporters_to_relay_v3( } +def test_finalize_hermes_relay_session_uses_legacy_plugin_hook(monkeypatch): + hermes_cli = ModuleType("hermes_cli") + hermes_plugins = ModuleType("hermes_cli.plugins") + mock_invoke_hook = MagicMock() + hermes_plugins.invoke_hook = mock_invoke_hook # type: ignore[attr-defined] + + monkeypatch.setitem(sys.modules, "hermes_cli", hermes_cli) + monkeypatch.setitem(sys.modules, "hermes_cli.plugins", hermes_plugins) + monkeypatch.delitem(sys.modules, "hermes_cli.lifecycle", raising=False) + + adapter.finalize_hermes_relay_session("session-legacy") + + mock_invoke_hook.assert_called_once_with( + "on_session_finalize", session_id="session-legacy", platform="fabric" + ) + + async def test_runtime_start_stages_upstream_relay_plugin_configuration( monkeypatch, tmp_path: Path, @@ -200,10 +219,21 @@ def stop_after_staging( ) -> tuple[Path, dict[str, object]]: assert relay_enabled is True assert os.environ["HERMES_NEMO_RELAY_PLUGINS_TOML"] == str(plugin_config_path) + assert all( + name not in os.environ + for name in adapter.HERMES_RELAY_ENV_NAMES + if name != "HERMES_NEMO_RELAY_PLUGINS_TOML" + ) raise RuntimeError("stop after Relay plugin staging") monkeypatch.setattr(adapter, "write_hermes_config", stop_after_staging) - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "before-start") + inherited_relay_environment = { + name: f"before-{index}" + for index, name in enumerate(adapter.HERMES_RELAY_ENV_NAMES) + if name != "HERMES_NEMO_RELAY_PLUGINS_TOML" + } + for name, value in inherited_relay_environment.items(): + monkeypatch.setenv(name, value) payload = { "base_dir": str(tmp_path), "config": { @@ -222,7 +252,9 @@ def stop_after_staging( await adapter.HermesRuntime().start(payload) assert "HERMES_NEMO_RELAY_PLUGINS_TOML" not in os.environ - assert os.environ["HERMES_NEMO_RELAY_ATIF_ENABLED"] == "before-start" + assert { + name: os.environ[name] for name in inherited_relay_environment + } == inherited_relay_environment def test_build_hermes_config_maps_fabric_config_to_hermes_config(): @@ -865,6 +897,59 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history( ) +async def test_runtime_stop_waits_for_cancelled_invoke_worker(monkeypatch): + worker_started = threading.Event() + worker_release = threading.Event() + mock_agent = MagicMock() + mock_session_db = MagicMock() + + def run_turn(**_kwargs): + worker_started.set() + assert worker_release.wait(timeout=1) + return ( + { + "response": "completed after cancellation", + "completed": True, + "failed": False, + "messages": [], + }, + "", + ) + + monkeypatch.setattr(adapter, "_invoke_hermes_turn", run_turn) + runtime = adapter.HermesRuntime() + runtime._started = True + runtime._runtime_id = "runtime-cancelled-invoke" + runtime._start_payload = {"config": {"instructions": {}}} + runtime._agent = mock_agent + runtime._session_db = mock_session_db + + invoke_task = asyncio.create_task( + runtime.invoke( + { + "runtime_context": {"runtime_id": runtime._runtime_id}, + "request": {"input": "wait"}, + } + ) + ) + assert await asyncio.to_thread(worker_started.wait, 1) + + invoke_task.cancel() + with pytest.raises(asyncio.CancelledError): + await invoke_task + + stop_task = asyncio.create_task(runtime.stop()) + await asyncio.sleep(0) + mock_agent.close.assert_not_called() + mock_session_db.close.assert_not_called() + + worker_release.set() + await stop_task + + mock_agent.close.assert_called_once_with() + mock_session_db.close.assert_called_once_with() + + def test_main_serves_persistent_runtime(monkeypatch): serve = MagicMock() monkeypatch.setattr(adapter.lifecycle, "serve", serve) diff --git a/tests/adapters/test_hermes_streaming.py b/tests/adapters/test_hermes_streaming.py index f9e5081f..04b36957 100644 --- a/tests/adapters/test_hermes_streaming.py +++ b/tests/adapters/test_hermes_streaming.py @@ -64,7 +64,6 @@ async def test_relay_invocation_scope_carries_fabric_request_id( model="test-model", platform="fabric", ) - runtime._invoke_hook = lambda *_args, **_kwargs: events.append("finalize") runtime._relay_plugin_config = relay_plugin_config runtime._hermes_home = tmp_path runtime._hermes_config_path = tmp_path / "config.yaml" @@ -83,6 +82,11 @@ def invoke_turn(**_kwargs: object): ) monkeypatch.setattr(adapter, "_invoke_hermes_turn", invoke_turn) + monkeypatch.setattr( + adapter, + "finalize_hermes_relay_session", + lambda _session_id: events.append("finalize"), + ) monkeypatch.setattr( adapter.common_utils, "collect_relay_artifacts", diff --git a/tests/e2e/test_hermes_e2e.py b/tests/e2e/test_hermes_e2e.py index 963fc8dd..a01b8598 100644 --- a/tests/e2e/test_hermes_e2e.py +++ b/tests/e2e/test_hermes_e2e.py @@ -7,9 +7,11 @@ import os import sys import warnings +from importlib.metadata import version as distribution_version from pathlib import Path from types import ModuleType +from packaging.version import Version import pytest import requests import yaml @@ -106,19 +108,19 @@ async def test_mcp_stdio_transport( ): os.environ["ADAPTER_PYTHON"] = sys.executable tool_name = "mcp__mcp_server_time__get_current_time" + tool_arguments = {"timezone": "America/Los_Angeles"} + if Version(distribution_version("hermes-agent")) < Version("0.20"): + # The released 0.19 integration accepts the configured MCP tool directly. + tool_call = {"name": tool_name, "arguments": tool_arguments} + else: + # Newer Hermes versions dispatch MCP schemas through their native bridge. + tool_call = { + "name": "tool_call", + "arguments": {"name": tool_name, "arguments": tool_arguments}, + } scenario_response = requests.post( f"{api_server}/_scenario", - json={ - "tool_call": { - # Hermes v0.20 defers MCP schemas behind its native tool-call - # bridge. The bridge dispatches this to the configured MCP tool. - "name": "tool_call", - "arguments": { - "name": tool_name, - "arguments": {"timezone": "America/Los_Angeles"}, - }, - } - }, + json={"tool_call": tool_call}, timeout=5, ) scenario_response.raise_for_status() @@ -188,7 +190,9 @@ async def test_mcp_stdio_transport( record for record in tool_records if record["scope_category"] == "end" ) assert tool_end["category"] == "tool" - assert tool_end["metadata"]["otel.status_code"] == "OK" + assert tool_end["metadata"].get("otel.status_code") == "OK" or ( + tool_end["metadata"].get("status") == "ok" + ) assert "America/Los_Angeles" in tool_end["data"] @@ -327,36 +331,38 @@ async def test_atof_artifacts(self): record_kinds = { (record["name"], record.get("scope_category")) for record in atof_records } - assert record_kinds.issuperset( - { - ("hermes.session", "start"), - ("hermes.session", "end"), - ("hermes.turn", "start"), - ("hermes.turn", "end"), - ("nvidia", "start"), - ("nvidia", "end"), - } - ) + assert record_kinds.issuperset({("nvidia", "start"), ("nvidia", "end")}) - fabric_scopes = [ + session_scopes = [ record for record in atof_records - if record["name"] in {"hermes.session", "hermes.turn"} + if record["name"] == "hermes.session" + or str(record["name"]).startswith("hermes-session-") ] - assert all( - record["metadata"]["hermes.execution_surface"] == "fabric" - for record in fabric_scopes - ) - turn_marks = [ + assert {record.get("scope_category") for record in session_scopes} >= { + "start", + "end", + } + + current_turn_scopes = [ + record + for record in atof_records + if record["name"] == "hermes.turn" + and record.get("scope_category") in {"start", "end"} + ] + legacy_turn_marks = [ record for record in atof_records if record["name"] in {"hermes.turn.start", "hermes.turn.end"} ] + turn_marks = current_turn_scopes or legacy_turn_marks + assert turn_marks + + fabric_scopes = [*session_scopes, *turn_marks] assert all( - record["metadata"]["model"] - == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" - and record["metadata"]["platform"] == self.atof_platform - for record in turn_marks + record["metadata"].get("hermes.execution_surface") == "fabric" + or record["metadata"].get("platform") == self.atof_platform + for record in fabric_scopes ) async def test_atif_artifacts(self): diff --git a/tests/python/test_streaming.py b/tests/python/test_streaming.py index 64a7fe1d..c16a4f49 100644 --- a/tests/python/test_streaming.py +++ b/tests/python/test_streaming.py @@ -931,6 +931,50 @@ async def test_listener_correlates_records_to_active_turn( await listener.close() +async def test_listener_correlates_legacy_hermes_turn_records(): + listener = await _AtofStreamListener(maxsize=4).start() + listener.begin_stream(request_id="request-2", turn_index=2) + turn_id = "legacy-turn" + current = [ + { + "kind": "mark", + "uuid": "turn-start", + "name": "hermes.turn.start", + "metadata": {"platform": "fabric", "turn_id": turn_id}, + }, + { + "kind": "scope", + "scope_category": "start", + "uuid": "llm", + "parent_uuid": "session", + "metadata": {"turn_id": turn_id}, + }, + { + "kind": "mark", + "uuid": "turn-end", + "name": "hermes.turn.end", + "metadata": {"platform": "fabric", "turn_id": turn_id}, + }, + ] + + await _post_chunked( + listener.url, + [ + { + "kind": "scope", + "scope_category": "start", + "uuid": "previous", + "metadata": {"nemo_fabric_request_id": "request-1"}, + }, + *current, + ], + ) + + assert [await listener.records.get() for _ in current] == current + listener.end_stream() + await listener.close() + + async def test_listener_applies_byte_budget_backpressure(): record = {"uuid": "record", "payload": "x" * 16} record_size = len(json.dumps(record).encode()) From 0170a88322194fea788b1d687398e90b40a15309 Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 14:08:19 -0700 Subject: [PATCH 06/12] fix(hermes): recover cancelled relay invocations Signed-off-by: Zhongxuan Wang --- .../nemo_fabric_adapters/hermes/adapter.py | 8 +++ python/src/nemo_fabric/streaming.py | 5 ++ tests/adapters/test_hermes_adapter.py | 54 +++++++++++++++++++ tests/python/test_streaming.py | 1 + 4 files changed, 68 insertions(+) diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index 9ccd1e62..a08ff7f5 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -518,6 +518,14 @@ def invoke_turn() -> tuple[dict[str, Any], str]: ) invoke_task = asyncio.create_task(asyncio.to_thread(invoke_turn)) self._active_invoke_task = invoke_task + + def clear_active_invoke_task( + completed_task: asyncio.Task[tuple[dict[str, Any], str]], + ) -> None: + if self._active_invoke_task is completed_task: + self._active_invoke_task = None + + invoke_task.add_done_callback(clear_active_invoke_task) try: result, adapter_stdout = await asyncio.shield(invoke_task) finally: diff --git a/python/src/nemo_fabric/streaming.py b/python/src/nemo_fabric/streaming.py index 8897d737..6bf1cc38 100644 --- a/python/src/nemo_fabric/streaming.py +++ b/python/src/nemo_fabric/streaming.py @@ -530,6 +530,11 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool: and isinstance(metadata, dict) and metadata.get("turn_id") == self._legacy_hermes_turn_id ): + if ( + record.get("kind") == "scope" + and record.get("scope_category") == "start" + ): + self._turn_scope_uuids.add(uuid) return True if self._turn_root_uuid is None: if not self._matches_turn_root(record): diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index a0d1baab..3459eab3 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -950,6 +950,60 @@ def run_turn(**_kwargs): mock_session_db.close.assert_called_once_with() +async def test_runtime_allows_invoke_after_cancelled_worker_finishes(monkeypatch): + worker_started = threading.Event() + worker_finished = threading.Event() + worker_release = threading.Event() + mock_agent = MagicMock() + mock_session_db = MagicMock() + calls = 0 + + def run_turn(**_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + worker_started.set() + assert worker_release.wait(timeout=1) + worker_finished.set() + return ( + { + "response": f"turn-{calls}", + "completed": True, + "failed": False, + "messages": [], + }, + "", + ) + + monkeypatch.setattr(adapter, "_invoke_hermes_turn", run_turn) + runtime = adapter.HermesRuntime() + runtime._started = True + runtime._runtime_id = "runtime-cancelled-invoke" + runtime._start_payload = {"config": {"instructions": {}}} + runtime._agent = mock_agent + runtime._session_db = mock_session_db + invocation = { + "runtime_context": {"runtime_id": runtime._runtime_id}, + "request": {"input": "wait"}, + } + + cancelled_invoke = asyncio.create_task(runtime.invoke(invocation)) + assert await asyncio.to_thread(worker_started.wait, 1) + + cancelled_invoke.cancel() + with pytest.raises(asyncio.CancelledError): + await cancelled_invoke + + worker_release.set() + assert await asyncio.to_thread(worker_finished.wait, 1) + await asyncio.sleep(0) + + result = await runtime.invoke(invocation) + + assert result["response"] == "turn-2" + await runtime.stop() + + def test_main_serves_persistent_runtime(monkeypatch): serve = MagicMock() monkeypatch.setattr(adapter.lifecycle, "serve", serve) diff --git a/tests/python/test_streaming.py b/tests/python/test_streaming.py index c16a4f49..266b1b9b 100644 --- a/tests/python/test_streaming.py +++ b/tests/python/test_streaming.py @@ -949,6 +949,7 @@ async def test_listener_correlates_legacy_hermes_turn_records(): "parent_uuid": "session", "metadata": {"turn_id": turn_id}, }, + {"kind": "mark", "uuid": "llm-child", "parent_uuid": "llm"}, { "kind": "mark", "uuid": "turn-end", From bafb5d7a93322e79999d4b163a29811665fb801a Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 15:45:46 -0700 Subject: [PATCH 07/12] test(hermes): wait for cancelled invoke completion Signed-off-by: Zhongxuan Wang --- tests/adapters/test_hermes_adapter.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index 3459eab3..4f96e8f9 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -952,7 +952,6 @@ def run_turn(**_kwargs): async def test_runtime_allows_invoke_after_cancelled_worker_finishes(monkeypatch): worker_started = threading.Event() - worker_finished = threading.Event() worker_release = threading.Event() mock_agent = MagicMock() mock_session_db = MagicMock() @@ -964,7 +963,6 @@ def run_turn(**_kwargs): if calls == 1: worker_started.set() assert worker_release.wait(timeout=1) - worker_finished.set() return ( { "response": f"turn-{calls}", @@ -989,14 +987,15 @@ def run_turn(**_kwargs): cancelled_invoke = asyncio.create_task(runtime.invoke(invocation)) assert await asyncio.to_thread(worker_started.wait, 1) + active_invoke_task = runtime._active_invoke_task + assert active_invoke_task is not None cancelled_invoke.cancel() with pytest.raises(asyncio.CancelledError): await cancelled_invoke worker_release.set() - assert await asyncio.to_thread(worker_finished.wait, 1) - await asyncio.sleep(0) + await asyncio.wait_for(asyncio.shield(active_invoke_task), timeout=1) result = await runtime.invoke(invocation) From 3099c13e39bd1a112ecac14f5e2834f814468620 Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 15:55:22 -0700 Subject: [PATCH 08/12] fix(hermes): isolate Relay fallback flags Signed-off-by: Zhongxuan Wang --- .../hermes/src/nemo_fabric_adapters/hermes/adapter.py | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index a08ff7f5..b3fdc91d 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -37,17 +37,12 @@ "openai": "OPENAI_API_KEY", "openrouter": "OPENROUTER_API_KEY", } -# Hermes still discovers its Relay plugin through the TOML-path environment -# variable. Keep the related fallback names here solely to clear and restore -# inherited process state; Fabric does not translate its Relay config into them. +# Hermes 0.16+ discovers Relay from this TOML path and falls back to direct +# ATIF/ATOF only when TOML initialization fails. Clear only those enable flags. HERMES_RELAY_ENV_NAMES = ( "HERMES_NEMO_RELAY_PLUGINS_TOML", "HERMES_NEMO_RELAY_ATIF_ENABLED", - "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", - "HERMES_NEMO_RELAY_ATIF_FILENAME_TEMPLATE", - "HERMES_NEMO_RELAY_ATIF_AGENT_NAME", - "HERMES_NEMO_RELAY_ATIF_AGENT_VERSION", - "HERMES_NEMO_RELAY_ATIF_MODEL_NAME", + "HERMES_NEMO_RELAY_ATOF_ENABLED", ) From 4dd81e75a65dec3ae583e7b6119df8e172d2b084 Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 16:10:38 -0700 Subject: [PATCH 09/12] refactor(streaming): remove Hermes correlation fallback Signed-off-by: Zhongxuan Wang --- python/src/nemo_fabric/streaming.py | 28 --------------- tests/python/test_streaming.py | 54 ++--------------------------- 2 files changed, 3 insertions(+), 79 deletions(-) diff --git a/python/src/nemo_fabric/streaming.py b/python/src/nemo_fabric/streaming.py index 6bf1cc38..df8b908c 100644 --- a/python/src/nemo_fabric/streaming.py +++ b/python/src/nemo_fabric/streaming.py @@ -255,7 +255,6 @@ def __init__( self._request_id: str | None = None self._turn_index: int | None = None self._turn_root_uuid: str | None = None - self._legacy_hermes_turn_id: str | None = None self._turn_scope_uuids: set[str] = set() self._saw_atof_data = False self._matched_turn_root = False @@ -309,7 +308,6 @@ def begin_stream( self._request_id = request_id self._turn_index = turn_index self._turn_root_uuid = None - self._legacy_hermes_turn_id = None self._turn_scope_uuids.clear() self._saw_atof_data = False self._matched_turn_root = request_id is None and turn_index is None @@ -324,7 +322,6 @@ def end_stream(self) -> None: self._request_id = None self._turn_index = None self._turn_root_uuid = None - self._legacy_hermes_turn_id = None self._turn_scope_uuids.clear() def warn_if_unavailable(self) -> None: @@ -525,25 +522,10 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool: if not isinstance(uuid, str): return False metadata = record.get("metadata") - if ( - self._legacy_hermes_turn_id is not None - and isinstance(metadata, dict) - and metadata.get("turn_id") == self._legacy_hermes_turn_id - ): - if ( - record.get("kind") == "scope" - and record.get("scope_category") == "start" - ): - self._turn_scope_uuids.add(uuid) - return True if self._turn_root_uuid is None: if not self._matches_turn_root(record): return False self._turn_root_uuid = uuid - if isinstance(metadata, dict) and record.get("kind") == "mark": - turn_id = metadata.get("turn_id") - if isinstance(turn_id, str): - self._legacy_hermes_turn_id = turn_id self._turn_scope_uuids.add(uuid) self._matched_turn_root = True return True @@ -564,13 +546,6 @@ def _matches_turn_root(self, record: dict[str, Any]) -> bool: metadata = record.get("metadata") if not isinstance(metadata, dict): return False - if ( - record.get("kind") == "mark" - and record.get("name") == "hermes.turn.start" - and metadata.get("platform") == "fabric" - and isinstance(metadata.get("turn_id"), str) - ): - return True if record.get("kind") != "scope" or record.get("scope_category") != "start": return False if ( @@ -582,9 +557,6 @@ def _matches_turn_root(self, record: dict[str, Any]) -> bool: self._turn_index is not None and metadata.get("nemo_relay_scope_role") == "turn" and metadata.get("turn_index") == self._turn_index - ) or ( - record.get("name") == "hermes.turn" - and metadata.get("hermes.execution_surface") == "fabric" ) async def close(self) -> None: diff --git a/tests/python/test_streaming.py b/tests/python/test_streaming.py index 266b1b9b..31736cec 100644 --- a/tests/python/test_streaming.py +++ b/tests/python/test_streaming.py @@ -880,16 +880,14 @@ async def produce() -> None: @pytest.mark.parametrize( - ("current_metadata", "current_name"), + "current_metadata", [ - ({"nemo_fabric_request_id": "request-2"}, None), - ({"nemo_relay_scope_role": "turn", "turn_index": 2}, None), - ({"hermes.execution_surface": "fabric"}, "hermes.turn"), + {"nemo_fabric_request_id": "request-2"}, + {"nemo_relay_scope_role": "turn", "turn_index": 2}, ], ) async def test_listener_correlates_records_to_active_turn( current_metadata: dict[str, Any], - current_name: str | None, ): listener = await _AtofStreamListener(maxsize=4).start() listener.begin_stream(request_id="request-2", turn_index=2) @@ -898,7 +896,6 @@ async def test_listener_correlates_records_to_active_turn( "kind": "scope", "scope_category": "start", "uuid": "current", - "name": current_name, "metadata": current_metadata, }, { @@ -931,51 +928,6 @@ async def test_listener_correlates_records_to_active_turn( await listener.close() -async def test_listener_correlates_legacy_hermes_turn_records(): - listener = await _AtofStreamListener(maxsize=4).start() - listener.begin_stream(request_id="request-2", turn_index=2) - turn_id = "legacy-turn" - current = [ - { - "kind": "mark", - "uuid": "turn-start", - "name": "hermes.turn.start", - "metadata": {"platform": "fabric", "turn_id": turn_id}, - }, - { - "kind": "scope", - "scope_category": "start", - "uuid": "llm", - "parent_uuid": "session", - "metadata": {"turn_id": turn_id}, - }, - {"kind": "mark", "uuid": "llm-child", "parent_uuid": "llm"}, - { - "kind": "mark", - "uuid": "turn-end", - "name": "hermes.turn.end", - "metadata": {"platform": "fabric", "turn_id": turn_id}, - }, - ] - - await _post_chunked( - listener.url, - [ - { - "kind": "scope", - "scope_category": "start", - "uuid": "previous", - "metadata": {"nemo_fabric_request_id": "request-1"}, - }, - *current, - ], - ) - - assert [await listener.records.get() for _ in current] == current - listener.end_stream() - await listener.close() - - async def test_listener_applies_byte_budget_backpressure(): record = {"uuid": "record", "payload": "x" * 16} record_size = len(json.dumps(record).encode()) From f9fedd05020900aa3cb45c7d8594a6a8967253e9 Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 16:13:26 -0700 Subject: [PATCH 10/12] refactor(hermes): own Relay environment lifecycle Signed-off-by: Zhongxuan Wang --- .../nemo_fabric_adapters/hermes/adapter.py | 36 ++++--------------- tests/adapters/test_hermes_adapter.py | 14 ++------ 2 files changed, 10 insertions(+), 40 deletions(-) diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index b3fdc91d..4c4c838c 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -322,8 +322,6 @@ def __init__(self) -> None: self._agent: Any = None self._relay_plugin_config: dict[str, Any] | None = None self._relay_plugin_config_path: Path | None = None - self._previous_relay_environment: dict[str, str | None] = {} - self._applied_relay_environment: dict[str, str | None] = {} self._active_invoke_task: asyncio.Task[tuple[dict[str, Any], str]] | None = None async def start(self, payload: dict[str, Any]) -> None: @@ -359,22 +357,11 @@ async def start(self, payload: dict[str, Any]) -> None: self._relay_plugin_config_path, self._relay_plugin_config, ) = write_hermes_relay_plugin_config(payload) - relay_environment = { - "HERMES_NEMO_RELAY_PLUGINS_TOML": str( - self._relay_plugin_config_path - ) - } - self._previous_relay_environment = { - name: os.environ.get(name) for name in HERMES_RELAY_ENV_NAMES - } - self._applied_relay_environment = {} for name in HERMES_RELAY_ENV_NAMES: - value = relay_environment.get(name) - if value is None: - os.environ.pop(name, None) - else: - os.environ[name] = value - self._applied_relay_environment[name] = value + os.environ.pop(name, None) + os.environ["HERMES_NEMO_RELAY_PLUGINS_TOML"] = str( + self._relay_plugin_config_path + ) self._hermes_config_path, self._hermes_config = write_hermes_config( payload, @@ -576,8 +563,7 @@ async def stop(self) -> None: agent = self._agent session_db = self._session_db had_mcp_servers = bool(self._hermes_config.get("mcp_servers")) - previous_relay_environment = self._previous_relay_environment - applied_relay_environment = self._applied_relay_environment + had_relay_plugin = self._relay_plugin_config_path is not None self._agent = None self._session_db = None self._start_payload = None @@ -592,19 +578,11 @@ async def stop(self) -> None: self._conversation_history = None self._relay_plugin_config = None self._relay_plugin_config_path = None - self._previous_relay_environment = {} - self._applied_relay_environment = {} self._started = False - for name, applied_value in applied_relay_environment.items(): - current_value = os.environ.get(name) - if current_value != applied_value: - continue - previous_value = previous_relay_environment.get(name) - if previous_value is None: + if had_relay_plugin: + for name in HERMES_RELAY_ENV_NAMES: os.environ.pop(name, None) - else: - os.environ[name] = previous_value if had_mcp_servers: try: diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index 4f96e8f9..fe61cbc7 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -227,13 +227,8 @@ def stop_after_staging( raise RuntimeError("stop after Relay plugin staging") monkeypatch.setattr(adapter, "write_hermes_config", stop_after_staging) - inherited_relay_environment = { - name: f"before-{index}" - for index, name in enumerate(adapter.HERMES_RELAY_ENV_NAMES) - if name != "HERMES_NEMO_RELAY_PLUGINS_TOML" - } - for name, value in inherited_relay_environment.items(): - monkeypatch.setenv(name, value) + for name in adapter.HERMES_RELAY_ENV_NAMES: + monkeypatch.setenv(name, "before") payload = { "base_dir": str(tmp_path), "config": { @@ -251,10 +246,7 @@ def stop_after_staging( with pytest.raises(RuntimeError, match="stop after Relay plugin staging"): await adapter.HermesRuntime().start(payload) - assert "HERMES_NEMO_RELAY_PLUGINS_TOML" not in os.environ - assert { - name: os.environ[name] for name in inherited_relay_environment - } == inherited_relay_environment + assert all(name not in os.environ for name in adapter.HERMES_RELAY_ENV_NAMES) def test_build_hermes_config_maps_fabric_config_to_hermes_config(): From 31b06eccbae2dd8cee3ab206fd8b764d25bed782 Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 16:23:31 -0700 Subject: [PATCH 11/12] fix(streaming): correlate upstream Hermes turns Signed-off-by: Zhongxuan Wang --- python/src/nemo_fabric/streaming.py | 29 ++++++++++++++++++ tests/e2e/test_hermes_e2e.py | 4 +-- tests/python/test_streaming.py | 46 +++++++++++++++++++++++++++++ 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/python/src/nemo_fabric/streaming.py b/python/src/nemo_fabric/streaming.py index df8b908c..aa5821b2 100644 --- a/python/src/nemo_fabric/streaming.py +++ b/python/src/nemo_fabric/streaming.py @@ -254,6 +254,7 @@ def __init__( self._accepting = False self._request_id: str | None = None self._turn_index: int | None = None + self._upstream_hermes_turn_id: str | None = None self._turn_root_uuid: str | None = None self._turn_scope_uuids: set[str] = set() self._saw_atof_data = False @@ -307,6 +308,7 @@ def begin_stream( self._queue.get_nowait() self._request_id = request_id self._turn_index = turn_index + self._upstream_hermes_turn_id = None self._turn_root_uuid = None self._turn_scope_uuids.clear() self._saw_atof_data = False @@ -321,6 +323,7 @@ def end_stream(self) -> None: self._accepting = False self._request_id = None self._turn_index = None + self._upstream_hermes_turn_id = None self._turn_root_uuid = None self._turn_scope_uuids.clear() @@ -522,9 +525,28 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool: if not isinstance(uuid, str): return False metadata = record.get("metadata") + + # Upstream Hermes emits its turn markers from its own Relay runtime, + # so they carry the upstream turn ID rather than Fabric scope metadata. + if ( + self._upstream_hermes_turn_id is not None + and isinstance(metadata, dict) + and metadata.get("turn_id") == self._upstream_hermes_turn_id + ): + if ( + record.get("kind") == "scope" + and record.get("scope_category") == "start" + ): + self._turn_scope_uuids.add(uuid) + return True + if self._turn_root_uuid is None: if not self._matches_turn_root(record): return False + if isinstance(metadata, dict) and record.get("kind") == "mark": + turn_id = metadata.get("turn_id") + if isinstance(turn_id, str): + self._upstream_hermes_turn_id = turn_id self._turn_root_uuid = uuid self._turn_scope_uuids.add(uuid) self._matched_turn_root = True @@ -546,6 +568,13 @@ def _matches_turn_root(self, record: dict[str, Any]) -> bool: metadata = record.get("metadata") if not isinstance(metadata, dict): return False + if ( + record.get("kind") == "mark" + and record.get("name") == "hermes.turn.start" + and metadata.get("platform") == "fabric" + and isinstance(metadata.get("turn_id"), str) + ): + return True if record.get("kind") != "scope" or record.get("scope_category") != "start": return False if ( diff --git a/tests/e2e/test_hermes_e2e.py b/tests/e2e/test_hermes_e2e.py index a01b8598..3ecf5ad2 100644 --- a/tests/e2e/test_hermes_e2e.py +++ b/tests/e2e/test_hermes_e2e.py @@ -350,12 +350,12 @@ async def test_atof_artifacts(self): if record["name"] == "hermes.turn" and record.get("scope_category") in {"start", "end"} ] - legacy_turn_marks = [ + upstream_turn_marks = [ record for record in atof_records if record["name"] in {"hermes.turn.start", "hermes.turn.end"} ] - turn_marks = current_turn_scopes or legacy_turn_marks + turn_marks = current_turn_scopes or upstream_turn_marks assert turn_marks fabric_scopes = [*session_scopes, *turn_marks] diff --git a/tests/python/test_streaming.py b/tests/python/test_streaming.py index 31736cec..88ee56c0 100644 --- a/tests/python/test_streaming.py +++ b/tests/python/test_streaming.py @@ -928,6 +928,52 @@ async def test_listener_correlates_records_to_active_turn( await listener.close() +async def test_listener_correlates_upstream_hermes_turn_records(): + listener = await _AtofStreamListener(maxsize=4).start() + listener.begin_stream(request_id="request-2", turn_index=2) + turn_id = "upstream-turn" + current = [ + { + "kind": "mark", + "name": "hermes.turn.start", + "uuid": "turn-start", + "metadata": {"platform": "fabric", "turn_id": turn_id}, + }, + { + "kind": "scope", + "scope_category": "start", + "uuid": "llm", + "parent_uuid": "session", + "metadata": {"turn_id": turn_id}, + }, + {"kind": "mark", "uuid": "llm-child", "parent_uuid": "llm"}, + { + "kind": "mark", + "name": "hermes.turn.end", + "uuid": "turn-end", + "metadata": {"platform": "fabric", "turn_id": turn_id}, + }, + ] + + await _post_chunked( + listener.url, + [ + { + "kind": "scope", + "scope_category": "start", + "uuid": "previous", + "metadata": {"nemo_fabric_request_id": "request-1"}, + }, + *current, + ], + ) + + assert [await listener.records.get() for _ in current] == current + assert listener.records.empty() + listener.end_stream() + await listener.close() + + async def test_listener_applies_byte_budget_backpressure(): record = {"uuid": "record", "payload": "x" * 16} record_size = len(json.dumps(record).encode()) From 20ea7e071c67730d79a69966ba06e0514def1beb Mon Sep 17 00:00:00 2001 From: Zhongxuan Wang Date: Mon, 10 Aug 2026 22:51:20 -0700 Subject: [PATCH 12/12] fix(hermes): preserve Relay artifact boundaries Signed-off-by: Zhongxuan Wang --- .../nemo_fabric_adapters/hermes/adapter.py | 42 +++---------- python/src/nemo_fabric/streaming.py | 5 +- tests/adapters/test_hermes_adapter.py | 4 ++ tests/adapters/test_hermes_streaming.py | 61 ++----------------- tests/e2e/test_hermes_e2e.py | 5 ++ tests/python/test_streaming.py | 48 ++++++++++++++- 6 files changed, 72 insertions(+), 93 deletions(-) diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py index 4c4c838c..75eb9951 100755 --- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py +++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py @@ -61,26 +61,6 @@ def finalize_hermes_relay_session(session_id: str) -> None: finalize_session(session_id=session_id, platform="fabric") -def _fabric_stream_sink_enabled(config: dict[str, Any] | None) -> bool: - if config is None: - return False - for component in config.get("components") or []: - if not isinstance(component, dict) or component.get("kind") != "observability": - continue - component_config = component.get("config") - if not isinstance(component_config, dict): - continue - atof = component_config.get("atof") - if not isinstance(atof, dict): - continue - if any( - isinstance(sink, dict) and sink.get("name") == "nemo-fabric-stream" - for sink in atof.get("sinks") or [] - ): - return True - return False - - def _api_key_env(model_config: dict[str, Any]) -> str: explicit = model_config.get("api_key_env") if isinstance(explicit, str) and explicit: @@ -467,6 +447,11 @@ def run_hermes_turn() -> tuple[dict[str, Any], str]: system_prompt=common_utils.system_instruction(start_payload), user_message=user_message, conversation_history=self._conversation_history, + task_id=( + request["request_id"] + if isinstance(request.get("request_id"), str) + else None + ), ) finally: if self._relay_plugin_config is not None: @@ -477,19 +462,6 @@ def run_hermes_turn() -> tuple[dict[str, Any], str]: # directly. finalize_hermes_relay_session(str(self._agent.session_id)) - def invoke_turn() -> tuple[dict[str, Any], str]: - if not _fabric_stream_sink_enabled(self._relay_plugin_config): - return run_hermes_turn() - - from nemo_relay import ScopeType, scope - - with scope.scope( - "nemo-fabric-invocation", - ScopeType.Agent, - metadata={"nemo_fabric_request_id": request.get("request_id")}, - ): - return run_hermes_turn() - # Hermes' upstream Relay integration drives async Relay hooks from its # synchronous agent loop. Run that loop outside this lifecycle server's # event-loop thread so Hermes can own its Relay event loop. @@ -498,7 +470,7 @@ def invoke_turn() -> tuple[dict[str, Any], str]: "hermes_invocation_in_progress", "Hermes runtime already has an active invocation", ) - invoke_task = asyncio.create_task(asyncio.to_thread(invoke_turn)) + invoke_task = asyncio.create_task(asyncio.to_thread(run_hermes_turn)) self._active_invoke_task = invoke_task def clear_active_invoke_task( @@ -623,6 +595,7 @@ def _invoke_hermes_turn( system_prompt: str | None, user_message: str, conversation_history: list[dict[str, Any]] | None, + task_id: str | None, ) -> tuple[dict[str, Any], str]: hermes_stdout = StringIO() with redirect_stdout(hermes_stdout): @@ -630,6 +603,7 @@ def _invoke_hermes_turn( agent.run_conversation, system_message=system_prompt, conversation_history=conversation_history, + task_id=task_id, sync_honcho=False, dont_review=True, ) diff --git a/python/src/nemo_fabric/streaming.py b/python/src/nemo_fabric/streaming.py index aa5821b2..5de3d397 100644 --- a/python/src/nemo_fabric/streaming.py +++ b/python/src/nemo_fabric/streaming.py @@ -526,8 +526,7 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool: return False metadata = record.get("metadata") - # Upstream Hermes emits its turn markers from its own Relay runtime, - # so they carry the upstream turn ID rather than Fabric scope metadata. + # Hermes copies the task ID passed by Fabric into its Relay turn markers. if ( self._upstream_hermes_turn_id is not None and isinstance(metadata, dict) @@ -572,6 +571,8 @@ def _matches_turn_root(self, record: dict[str, Any]) -> bool: record.get("kind") == "mark" and record.get("name") == "hermes.turn.start" and metadata.get("platform") == "fabric" + and self._request_id is not None + and metadata.get("task_id") == self._request_id and isinstance(metadata.get("turn_id"), str) ): return True diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py index fe61cbc7..e8b340e3 100644 --- a/tests/adapters/test_hermes_adapter.py +++ b/tests/adapters/test_hermes_adapter.py @@ -812,6 +812,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history( }, "request": { "input": "hello", + "request_id": "request-1", "context": {"history": [{"role": "user", "content": "stale"}]}, }, "capability_plan": {"native": {}}, @@ -829,6 +830,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history( ) payload["runtime_context"]["invocation_id"] = "invocation-2" payload["request"]["input"] = "continue" + payload["request"]["request_id"] = "request-2" second = await runtime.invoke( { "runtime_context": payload["runtime_context"], @@ -863,11 +865,13 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history( assert first_call.kwargs == { "system_message": "system", "conversation_history": None, + "task_id": "request-1", } assert second_call.args == ("continue",) assert second_call.kwargs == { "system_message": "system", "conversation_history": first_messages, + "task_id": "request-2", } mock_ai_agent.close.assert_called_once_with() mock_session_db.close.assert_called_once_with() diff --git a/tests/adapters/test_hermes_streaming.py b/tests/adapters/test_hermes_streaming.py index 04b36957..4aa0c672 100644 --- a/tests/adapters/test_hermes_streaming.py +++ b/tests/adapters/test_hermes_streaming.py @@ -4,7 +4,6 @@ """Dependency-free tests for Hermes Relay streaming integration.""" import sys -from contextlib import contextmanager from pathlib import Path from types import SimpleNamespace @@ -19,42 +18,12 @@ from nemo_fabric_adapters.hermes import adapter -@pytest.mark.parametrize( - ("relay_plugin_config", "expected_metadata"), - [ - ( - { - "components": [ - { - "kind": "observability", - "config": { - "atof": { - "enabled": True, - "sinks": [ - { - "type": "stream", - "name": "nemo-fabric-stream", - "url": "http://127.0.0.1:1234/atof", - } - ], - } - }, - } - ] - }, - [{"nemo_fabric_request_id": "request-1"}], - ), - ({"components": []}, []), - ], - ids=["streaming", "non-streaming"], -) -async def test_relay_invocation_scope_carries_fabric_request_id( +async def test_relay_invocation_passes_fabric_request_id_to_hermes( monkeypatch, tmp_path: Path, - relay_plugin_config: dict[str, object], - expected_metadata: list[object], ): events: list[str] = [] + task_ids: list[object] = [] runtime = adapter.HermesRuntime() runtime._started = True runtime._start_payload = {} @@ -64,13 +33,14 @@ async def test_relay_invocation_scope_carries_fabric_request_id( model="test-model", platform="fabric", ) - runtime._relay_plugin_config = relay_plugin_config + runtime._relay_plugin_config = {"components": []} runtime._hermes_home = tmp_path runtime._hermes_config_path = tmp_path / "config.yaml" runtime._enabled_toolsets = [] def invoke_turn(**_kwargs: object): events.append("turn") + task_ids.append(_kwargs["task_id"]) return ( { "response": "done", @@ -93,22 +63,6 @@ def invoke_turn(**_kwargs: object): lambda _config: [], ) - from nemo_relay import scope, subscribers - - captured_metadata: list[object] = [] - - @contextmanager - def capture_scope(*_args: object, **kwargs: object): - captured_metadata.append(kwargs["metadata"]) - events.append("scope-enter") - try: - yield - finally: - events.append("scope-exit") - - monkeypatch.setattr(scope, "scope", capture_scope) - monkeypatch.setattr(subscribers, "flush", lambda: None) - await runtime.invoke( { "runtime_context": {"runtime_id": "runtime-1"}, @@ -116,8 +70,5 @@ def capture_scope(*_args: object, **kwargs: object): } ) - assert captured_metadata == expected_metadata - if expected_metadata: - assert events == ["scope-enter", "turn", "finalize", "scope-exit"] - else: - assert events == ["turn", "finalize"] + assert task_ids == ["request-1"] + assert events == ["turn", "finalize"] diff --git a/tests/e2e/test_hermes_e2e.py b/tests/e2e/test_hermes_e2e.py index 3ecf5ad2..6a630bde 100644 --- a/tests/e2e/test_hermes_e2e.py +++ b/tests/e2e/test_hermes_e2e.py @@ -60,6 +60,11 @@ async def test_hermes_persistent_host_reuses_native_session( assert first["metadata"]["adapter_runner"] == "persistent_local_host", results assert first["metadata"]["host_pid"] == second["metadata"]["host_pid"], results assert "user_count=2" in second["output"]["response"], results + for turn in (first, second): + assert {artifact["kind"] for artifact in turn["output"]["relay_artifacts"]} >= { + "atof", + "atif", + }, turn.to_mapping() @pytest.mark.usefixtures("mock_nvidia_api_key", "nemo_relay") diff --git a/tests/python/test_streaming.py b/tests/python/test_streaming.py index 88ee56c0..9adf8236 100644 --- a/tests/python/test_streaming.py +++ b/tests/python/test_streaming.py @@ -937,7 +937,11 @@ async def test_listener_correlates_upstream_hermes_turn_records(): "kind": "mark", "name": "hermes.turn.start", "uuid": "turn-start", - "metadata": {"platform": "fabric", "turn_id": turn_id}, + "metadata": { + "platform": "fabric", + "task_id": "request-2", + "turn_id": turn_id, + }, }, { "kind": "scope", @@ -951,7 +955,11 @@ async def test_listener_correlates_upstream_hermes_turn_records(): "kind": "mark", "name": "hermes.turn.end", "uuid": "turn-end", - "metadata": {"platform": "fabric", "turn_id": turn_id}, + "metadata": { + "platform": "fabric", + "task_id": "request-2", + "turn_id": turn_id, + }, }, ] @@ -974,6 +982,42 @@ async def test_listener_correlates_upstream_hermes_turn_records(): await listener.close() +async def test_listener_rejects_late_upstream_hermes_turn_marker(): + listener = await _AtofStreamListener().start() + previous = { + "kind": "mark", + "name": "hermes.turn.start", + "uuid": "previous-turn", + "metadata": { + "platform": "fabric", + "task_id": "request-1", + "turn_id": "previous-turn", + }, + } + listener.begin_stream(request_id="request-1") + await _post_chunked(listener.url, [previous]) + assert await listener.records.get() == previous + listener.end_stream() + + current = { + "kind": "mark", + "name": "hermes.turn.start", + "uuid": "current-turn", + "metadata": { + "platform": "fabric", + "task_id": "request-2", + "turn_id": "current-turn", + }, + } + listener.begin_stream(request_id="request-2") + await _post_chunked(listener.url, [previous, current]) + + assert await listener.records.get() == current + assert listener.records.empty() + listener.end_stream() + await listener.close() + + async def test_listener_applies_byte_budget_backpressure(): record = {"uuid": "record", "payload": "x" * 16} record_size = len(json.dumps(record).encode())