mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-29 18:46:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(runtime): preserve provider payloads through Relay

Browse source 
Signed-off-by: Alex Fournier <afournier@nvidia.com>
This commit is contained in:
Alex Fournier 2026-07-21 09:25:22 -07:00
parent 67863777ca
commit 6efc5fe0af
12 changed files with 274 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

10
agent/relay_llm.py
View file

@ -32,7 +32,7 @@ def execute(
) -> Any:
"""Run one non-streaming physical provider attempt through Relay."""
runtime, session, parent = relay_runtime.resolve_execution_context(session_id)
if runtime is None or session is None:
if runtime is None or session is None or not runtime.managed_execution_enabled():
return callback(request)
logical = _logical_parent(runtime, session, parent, metadata)
parent = logical[1] if logical is not None else parent
@ -101,7 +101,7 @@ async def execute_async(
) -> Any:
"""Run one asynchronous physical provider attempt through Relay."""
runtime, session, parent = relay_runtime.resolve_execution_context(session_id)
if runtime is None or session is None:
if runtime is None or session is None or not runtime.managed_execution_enabled():
return await callback(request)
logical = _logical_parent(runtime, session, parent, metadata)
parent = logical[1] if logical is not None else parent
@ -299,7 +299,11 @@ class ManagedLlmStream(Iterator[Any]):
self.output_modified = False
runtime, session, parent = relay_runtime.resolve_execution_context(session_id)
if runtime is None or session is None:
if (
runtime is None
or session is None
or not runtime.managed_execution_enabled()
):
raw_stream = stream_factory(request)
if completed_response_predicate is not None and completed_response_predicate(
raw_stream

33
agent/relay_runtime.py
View file

@ -48,9 +48,28 @@ class RelayRuntime:
self._sessions: dict[str, RelaySession] = {}
self._subagent_parents: dict[str, str] = {}
self._subagent_parent_handles: dict[str, Any] = {}
self._execution_consumers_lock = threading.RLock()
self._execution_consumers: set[str] = set()
self._shutdown_registered = True
atexit.register(self.shutdown)
def retain_managed_execution(self, consumer: str) -> None:
"""Keep managed LLM and tool execution active for one consumer."""
if not consumer:
raise ValueError("Relay managed-execution consumer must not be empty")
with self._execution_consumers_lock:
self._execution_consumers.add(consumer)
def release_managed_execution(self, consumer: str) -> None:
"""Release a consumer's managed-execution requirement."""
with self._execution_consumers_lock:
self._execution_consumers.discard(consumer)
def managed_execution_enabled(self) -> bool:
"""Return whether an active interceptor or subscriber needs the pipeline."""
with self._execution_consumers_lock:
return bool(self._execution_consumers)
def ensure_session(
self,
event: dict[str, Any],
@ -247,6 +266,8 @@ class RelayRuntime:
args: dict[str, Any],
) -> dict[str, Any]:
"""Apply Relay request rewriting before Hermes authorizes a tool call."""
if not self.managed_execution_enabled():
return args
request_intercepts = getattr(
getattr(self.relay, "tools", None),
"request_intercepts",
@ -354,6 +375,18 @@ class NoopRelayRuntime:
del session_id, tool_name
return args
@staticmethod
def retain_managed_execution(consumer: str) -> None:
del consumer
@staticmethod
def release_managed_execution(consumer: str) -> None:
del consumer
@staticmethod
def managed_execution_enabled() -> bool:
return False
def shutdown(self) -> None:
"""No resources are allocated on unsupported platforms."""

2
agent/relay_tools.py
View file

@ -21,7 +21,7 @@ def execute(
) -> tuple[Any, dict[str, Any]]:
"""Run one tool call through Relay and return its final arguments."""
runtime, session, parent = relay_runtime.resolve_execution_context(session_id)
if runtime is None or session is None:
if runtime is None or session is None or not runtime.managed_execution_enabled():
return callback(args), args
observed_args = args

4
docs/observability/relay-shared-metrics.md
View file

@ -9,6 +9,10 @@ Hermes execution remains available, while Relay scopes, middleware, plugins,
and subscribers are unavailable. The `hermes-agent[nemo-relay]` extra remains
as a no-op compatibility alias for existing installation commands.
Hermes requires NeMo Relay 0.6 RC 3 or later. That release establishes the
lossless provider-codec contract used for Anthropic Messages, OpenAI Chat
Completions, and OpenAI Responses requests.
Collection remains off unless Hermes policy enables it:
```yaml

3
hermes_cli/observability/relay_shared_metrics.py
View file

@ -108,6 +108,7 @@ class _Runtime:
runtime_id=self.host.runtime_id,
)
self.relay.subscribers.register(self._subscriber_name, self.subscriber)
self.host.retain_managed_execution(self._subscriber_name)
self._registered = True
atexit.register(self.shutdown)
@ -401,6 +402,7 @@ class _Runtime:
self._safe(self.relay.subscribers.flush)
self._export()
self._safe(self.relay.subscribers.deregister, self._subscriber_name)
self.host.release_managed_execution(self._subscriber_name)
self._registered = False
try:
atexit.unregister(self.shutdown)
@ -414,6 +416,7 @@ class _Runtime:
self.subscriber.deactivate()
if self._registered:
self._safe(self.relay.subscribers.deregister, self._subscriber_name)
self.host.release_managed_execution(self._subscriber_name)
self._registered = False
with self._sessions_lock:
sessions = list(self._sessions.values())

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

@ -188,8 +188,8 @@ execution boundary.
### Dynamic Plugins
Hermes feature-detects the dynamic-plugin activation API available in NeMo Relay
0.6 and later. Configure native or worker plugins with Hermes-owned
Hermes uses the dynamic-plugin activation API available in NeMo Relay 0.6 and
later. Configure native or worker plugins with Hermes-owned
`[[dynamic_plugins]]` entries that match the Python binding's activation-spec
fields:
@ -238,12 +238,6 @@ During shutdown it closes session exporters, flushes Relay subscribers, and
then closes the activation so callbacks are removed before plugin code is
unloaded.
NeMo Relay 0.5 does not expose dynamic activation through its Python binding.
When dynamic plugin configuration is present with a binding that lacks the
activation API, Hermes logs an actionable warning and continues with the
ordinary static component configuration, so ATOF and ATIF observability remain
available. No dynamic plugin is loaded in that degraded mode.
For the full generic Hermes middleware contract, see
[`docs/middleware/README.md`](../../../docs/middleware/README.md).
@ -492,8 +486,8 @@ produces one Relay lifecycle.
This example enables both NeMo Relay observability export and adaptive execution
middleware for a local Hermes run. This path requires a NeMo Relay runtime that
supports `[components.config.tool_parallelism]`, as provided by the supported
0.x release range beginning with 0.5.
supports `[components.config.tool_parallelism]`, as provided by NeMo Relay 0.6
and later.
```bash
export HERMES_HOME=/tmp/hermes-middleware-test/hermes-home

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

@ -75,12 +75,30 @@ class _Runtime:
self.subagent_contexts: dict[str, _SubagentContext] = {}
self.atof_exporter: Any = None
self._atof_subscriber_name = f"hermes.nemo_relay.atof.{self.host.runtime_id}"
self._execution_consumer_name = (
f"hermes.nemo_relay.rich_observability.{self.host.runtime_id}"
)
self._execution_consumer_retained = False
self._plugin_activation: Any = None
self._shutdown_registered = False
self._plugin_config_initialized = self._configure_plugins_toml()
self._plugin_config_needs_reinit = False
if not self._plugin_config_initialized:
self._activate_direct_fallbacks()
self._sync_managed_execution()
def _sync_managed_execution(self) -> None:
required = bool(
self._plugin_config_initialized
or self.atof_exporter is not None
or self.settings.atif_enabled
)
if required and not self._execution_consumer_retained:
self.host.retain_managed_execution(self._execution_consumer_name)
self._execution_consumer_retained = True
elif not required and self._execution_consumer_retained:
self.host.release_managed_execution(self._execution_consumer_name)
self._execution_consumer_retained = False
def _configure_plugins_toml(self) -> bool:
if not self.settings.plugins_config:
@ -179,9 +197,11 @@ class _Runtime:
self._plugin_config_initialized = self._configure_plugins_toml()
if not self._plugin_config_initialized:
self._activate_direct_fallbacks()
self._sync_managed_execution()
return
self._clear_atof()
self._plugin_config_needs_reinit = False
self._sync_managed_execution()
def _plugins_toml_owns_exporter(self, exporter_name: str) -> bool:
return self._plugin_config_initialized and _observability_exporter_enabled(
@ -233,6 +253,7 @@ class _Runtime:
except Exception:
logger.debug("NeMo Relay ATOF deregister failed", exc_info=True)
self.atof_exporter = None
self._sync_managed_execution()
def prepare_session(self, kwargs: dict[str, Any]) -> _SessionState:
"""Register per-session subscribers without opening the core scope."""
@ -380,6 +401,9 @@ class _Runtime:
except Exception as exc:
failures.append(f"plugin runtime close failed: {exc}")
self._clear_atof()
if self._execution_consumer_retained:
self.host.release_managed_execution(self._execution_consumer_name)
self._execution_consumer_retained = False
if self._shutdown_registered and self._plugin_activation is None:
atexit.unregister(self.shutdown)
self._shutdown_registered = False

8
pyproject.toml
View file

@ -138,10 +138,10 @@ dependencies = [
# Hence the ``sys_platform == 'win32'`` marker: the dep (and its portalocker
# / pywin32 tree) ships only where it's actually used.
"concurrent-log-handler==0.9.29; sys_platform == 'win32'",
# First-party lifecycle and shared-metrics runtime. Follow the contribution
# policy's bounded pre-1.0 range while supporting Relay 0.5 and 0.6. Native
# wheels target these platforms; keep unsupported Hermes targets installable.
"nemo-relay>=0.5,<0.7; (sys_platform == 'darwin' and platform_machine == 'arm64') or (sys_platform == 'linux' and platform_machine == 'x86_64') or (sys_platform == 'linux' and platform_machine == 'aarch64') or (sys_platform == 'win32' and platform_machine == 'AMD64') or (sys_platform == 'win32' and platform_machine == 'ARM64')",
# First-party lifecycle and shared-metrics runtime. Relay 0.6 RC 3 is the
# minimum lossless provider-codec contract; final 0.6 releases also satisfy
# this bounded pre-1.0 range. Native wheels target these platforms.
"nemo-relay>=0.6.0rc3,<0.7; (sys_platform == 'darwin' and platform_machine == 'arm64') or (sys_platform == 'linux' and platform_machine == 'x86_64') or (sys_platform == 'linux' and platform_machine == 'aarch64') or (sys_platform == 'win32' and platform_machine == 'AMD64') or (sys_platform == 'win32' and platform_machine == 'ARM64')",
]
[project.optional-dependencies]

130
tests/agent/test_relay_llm.py
View file

@ -3,6 +3,7 @@
from __future__ import annotations
import contextvars
import json
from types import SimpleNamespace
import pytest
@ -24,9 +25,11 @@ def relay_turn(tmp_path, monkeypatch):
turn_id="turn-1",
task_id="task-1",
)
lease.host.retain_managed_execution("test.relay_llm")
try:
yield lease.host.relay, turn
finally:
lease.host.release_managed_execution("test.relay_llm")
relay_runtime.SESSION_COORDINATOR.end_turn(turn, outcome="success")
relay_runtime.SESSION_COORDINATOR.release_conversation(lease)
relay_runtime._reset_for_tests()
@ -325,6 +328,133 @@ def test_current_attempt_bypasses_relay_without_an_active_turn(monkeypatch):
assert result is request
def test_non_stream_bypasses_relay_without_an_active_consumer(relay_turn, monkeypatch):
relay, turn = relay_turn
turn.lease.host.release_managed_execution("test.relay_llm")
request = {"model": "test-model", "messages": [{"role": "user", "content": "hi"}]}
monkeypatch.setattr(
relay.llm,
"execute",
lambda *_args, **_kwargs: (_ for _ in ()).throw(
AssertionError("inactive Relay must not manage the provider call")
),
)
result = relay_llm.execute(
request,
lambda value: value,
session_id="session-1",
name="test-provider",
model_name="test-model",
)
assert result is request
def test_stream_bypasses_relay_without_an_active_consumer(relay_turn, monkeypatch):
relay, turn = relay_turn
turn.lease.host.release_managed_execution("test.relay_llm")
request = {"model": "test-model", "messages": [{"role": "user", "content": "hi"}]}
observed = []
monkeypatch.setattr(
relay.llm,
"stream_execute",
lambda *_args, **_kwargs: (_ for _ in ()).throw(
AssertionError("inactive Relay must not manage the provider stream")
),
)
stream = relay_llm.stream(
request,
lambda value: (observed.append(value), iter([{"delta": "ok"}]))[1],
session_id="session-1",
name="test-provider",
model_name="test-model",
finalizer=dict,
)
assert list(stream) == [{"delta": "ok"}]
assert observed == [request]
def test_anthropic_codec_preserves_tool_history_and_cached_system_blocks(relay_turn):
_relay, _turn = relay_turn
request = {
"model": "claude-sonnet-4-5",
"max_tokens": 512,
"system": [
{
"type": "text",
"text": "You are Hermes.",
"cache_control": {"type": "ephemeral"},
}
],
"messages": [
{"role": "user", "content": [{"type": "text", "text": "Run pwd"}]},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "toolu_01",
"name": "terminal",
"input": {"command": "pwd"},
}
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_01",
"content": [{"type": "text", "text": "/tmp/worktree"}],
}
],
},
],
}
original_wire = json.dumps(request, ensure_ascii=False, separators=(",", ":"))
observed_body_wire = ""
def provider(final_request):
nonlocal observed_body_wire
provider_body = {
key: value for key, value in final_request.items() if key != "extra_headers"
}
observed_body_wire = json.dumps(
provider_body,
ensure_ascii=False,
separators=(",", ":"),
)
return {
"id": "msg_01",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-5",
"content": [{"type": "text", "text": "Done"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 10, "output_tokens": 1},
}
relay_llm.execute(
request,
provider,
session_id="session-1",
name="anthropic",
model_name="claude-sonnet-4-5",
metadata={
"api_mode": "anthropic_messages",
"api_request_id": "request-anthropic",
},
)
assert observed_body_wire == original_wire
def test_current_attempt_bypasses_a_closed_turn_from_a_copied_context(
relay_turn,
monkeypatch,

56
tests/agent/test_relay_tools.py
View file

@ -21,14 +21,70 @@ def relay_turn(tmp_path, monkeypatch):
turn_id="turn-1",
task_id="task-1",
)
lease.host.retain_managed_execution("test.relay_tools")
try:
yield lease.host.relay
finally:
lease.host.release_managed_execution("test.relay_tools")
relay_runtime.SESSION_COORDINATOR.end_turn(turn, outcome="success")
relay_runtime.SESSION_COORDINATOR.release_conversation(lease)
relay_runtime._reset_for_tests()
def test_tool_adapter_bypasses_relay_without_an_active_consumer(
relay_turn, monkeypatch
):
relay = relay_turn
runtime = relay_runtime.get_runtime()
assert runtime is not None
runtime.release_managed_execution("test.relay_tools")
args = {"command": "pwd"}
monkeypatch.setattr(
relay.tools,
"execute",
lambda *_args, **_kwargs: (_ for _ in ()).throw(
AssertionError("inactive Relay must not manage the tool call")
),
)
result, final_args = relay_tools.execute(
"terminal",
args,
lambda value: value,
session_id="session-1",
)
assert result is args
assert final_args is args
def test_tool_request_intercepts_bypass_relay_without_an_active_consumer(
relay_turn, monkeypatch
):
relay = relay_turn
runtime = relay_runtime.get_runtime()
assert runtime is not None
runtime.release_managed_execution("test.relay_tools")
args = {"command": "pwd"}
monkeypatch.setattr(
relay.tools,
"request_intercepts",
lambda *_args, **_kwargs: (_ for _ in ()).throw(
AssertionError("inactive Relay must not run tool request intercepts")
),
)
final_args = runtime.apply_tool_request_intercepts(
session_id="session-1",
tool_name="terminal",
args=args,
)
assert final_args is args
def test_request_rewrite_reaches_authorized_callback_once(relay_turn):
relay = relay_turn
callback_args = []

2
tests/test_project_metadata.py
View file

@ -240,7 +240,7 @@ def test_nemo_relay_is_a_bounded_core_dependency():
]
assert len(relay_dependencies) == 1
requirement = Requirement(relay_dependencies[0])
assert str(requirement.specifier) == "<0.7,>=0.5"
assert str(requirement.specifier) == "<0.7,>=0.6.0rc3"
assert requirement.marker is not None
assert requirement.marker.evaluate(
{"sys_platform": "darwin", "platform_machine": "arm64"}

14
uv.lock generated
View file

@ -1802,7 +1802,7 @@ requires-dist = [
{ name = "microsoft-teams-apps", marker = "extra == 'teams'", specifier = "==2.0.13.4" },
{ name = "mistralai", marker = "extra == 'mistral'", specifier = "==2.4.8" },
{ name = "modal", marker = "extra == 'modal'", specifier = "==1.3.4" },
{ name = "nemo-relay", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux') or (platform_machine == 'AMD64' and sys_platform == 'win32') or (platform_machine == 'ARM64' and sys_platform == 'win32')", specifier = ">=0.5,<0.7" },
{ name = "nemo-relay", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux') or (platform_machine == 'AMD64' and sys_platform == 'win32') or (platform_machine == 'ARM64' and sys_platform == 'win32')", specifier = ">=0.6.0rc3,<0.7" },
{ name = "numpy", marker = "extra == 'voice'", specifier = "==2.4.3" },
{ name = "openai", specifier = "==2.24.0" },
{ name = "packaging", specifier = "==26.0" },
@ -2606,14 +2606,14 @@ wheels = [
[[package]]
name = "nemo-relay"
version = "0.5.0"
version = "0.6.0rc3"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/aa/fc/6892859ffbfb60f8cc09a4181c8d407b574d49490e35cd563e7f0a61bf54/nemo_relay-0.5.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:203409e0f392fc4f40f52277347ed75ba9969b2af28ddb3b6ae53f789185a0ed", size = 7769721, upload-time = "2026-07-08T18:43:57.745Z" },
{ url = "https://files.pythonhosted.org/packages/dd/13/25c4fcf6fb979af07c9562238819183588a910c2ae09767603500d6948c2/nemo_relay-0.5.0-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:de61ad0644c5e8c300de814f6884c5de7ae51db14694f5057f938b4da54bfaaf", size = 7037083, upload-time = "2026-07-08T18:44:00.135Z" },
{ url = "https://files.pythonhosted.org/packages/49/97/bd4c77e975d50f7d09f8bfcb74d7e704e05a8b2205f555bf557f84edc5fe/nemo_relay-0.5.0-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:cb323569f104506f4f1a0e2104b22d74c9a0bbf3280ec4d58ad3d768de76fb15", size = 7430817, upload-time = "2026-07-08T18:44:02.165Z" },
{ url = "https://files.pythonhosted.org/packages/da/8b/5681f3430e6d25bf7419dbe576da14599e1cfadc083edb74e691a0c3a980/nemo_relay-0.5.0-cp311-abi3-win_amd64.whl", hash = "sha256:da1bd1e1dec79b7bda3984dd8c59b32ba97abf96c37e85cd1f38d208a29f42b9", size = 7359048, upload-time = "2026-07-08T18:44:04.169Z" },
{ url = "https://files.pythonhosted.org/packages/4e/f2/65c3ef0435e671c829063d4872897f9054f369f0aeacab12c5aefe486705/nemo_relay-0.5.0-cp311-abi3-win_arm64.whl", hash = "sha256:87462585f17b40ce82c54347acef5ace96ee11de3015f234e0b902f23b3793fb", size = 6934235, upload-time = "2026-07-08T18:44:06.078Z" },
{ url = "https://files.pythonhosted.org/packages/d4/6a/abc154857deb7a62f21ed0eef0c989941690512597cb44928c5100d2858c/nemo_relay-0.6.0rc3-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:733a50ac46ae091434a729d9a590874379ee6a6acff4ef931dc7856c5c96aba4", size = 9895763, upload-time = "2026-07-21T06:07:48.838Z" },
{ url = "https://files.pythonhosted.org/packages/41/61/da06f03a0dda13252719f235b4b4ca79f3b0cc27ac62a86c6f3c7297002e/nemo_relay-0.6.0rc3-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7742c5348bbe71756441019a1063e4969060202fab4d1cba142a63a18c23d252", size = 8882012, upload-time = "2026-07-21T06:07:51.705Z" },
{ url = "https://files.pythonhosted.org/packages/a2/16/7fc4a6bf7f5f27305e73790b7eb351bc321d2dd9f9af3e01d3a5f40eadfe/nemo_relay-0.6.0rc3-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:98cff44f844f3bd45824b2a6ecc4fd571b521e94ec6b7fae1aca0b02c5384d89", size = 9322992, upload-time = "2026-07-21T06:07:53.767Z" },
{ url = "https://files.pythonhosted.org/packages/5d/12/57b0a58f578065226c513c0229ee78c26b3efd0ed6fa46c8d6ee2b949d94/nemo_relay-0.6.0rc3-cp311-abi3-win_amd64.whl", hash = "sha256:c8ad58c6cf05c74b667b1108609c2423bb00244b64a318a153b4aba91abf2fed", size = 9587358, upload-time = "2026-07-21T06:07:56.087Z" },
{ url = "https://files.pythonhosted.org/packages/8f/f9/63e41a74d3f2a493730158de0c2103f5c988f602d1f844f5daeed456078d/nemo_relay-0.6.0rc3-cp311-abi3-win_arm64.whl", hash = "sha256:934078a342c21bd94f3418f1551487e16616de09703229326d86341de0407187", size = 9019017, upload-time = "2026-07-21T06:07:58.072Z" },
]
[[package]]