hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity

refactor(relay): require 0.7 plugin APIs

Browse Source 
Signed-off-by: Bryan Bednarski <bbednarski@nvidia.com>
This commit is contained in:
Bryan Bednarski
2026-08-10 19:05:14 -06:00
parent 4e34dfc932
commit c86fd74e28
2 changed files with 40 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
agent/relay_runtime.py
View File

@@ -266,49 +266,34 @@ class _ProcessRelayPluginConfiguration:
return False
try:
plugin_mod = getattr(relay, "plugin", None)
configured_inputs = _configured_plugin_inputs(relay)
if configured_inputs is None:
return False
plugin_config, dynamic_plugins = configured_inputs
if dynamic_plugins:
initialize_dynamic = getattr(
plugin_mod,
"initialize_with_dynamic_plugins",
None,
)
if callable(initialize_dynamic):
try:
activation = _resolve_plugin_awaitable(
initialize_dynamic(plugin_config, dynamic_plugins)
try:
activation = _resolve_plugin_awaitable(
relay.plugin.initialize_with_dynamic_plugins(
plugin_config,
dynamic_plugins,
)
if activation is None:
raise RuntimeError(
"NeMo Relay dynamic plugin initialization "
"returned no activation handle"
)
self._activation = activation
except Exception as exc:
raise RuntimeError(
"Hermes Relay dynamic plugin activation failed"
) from exc
else:
raise RuntimeError(
"Hermes Relay dynamic plugins require a binding that "
"exposes plugin.initialize_with_dynamic_plugins"
)
if activation is None:
raise RuntimeError(
"NeMo Relay dynamic plugin initialization "
"returned no activation handle"
)
self._activation = activation
except Exception as exc:
raise RuntimeError(
"Hermes Relay dynamic plugin activation failed"
) from exc
if self._activation is None:
initialize = getattr(plugin_mod, "initialize", None)
if not callable(initialize):
raise RuntimeError(
"installed NeMo Relay binding does not expose "
"plugin.initialize"
)
# Hermes only enters Relay's initialization path after an
# explicit opt-in. Relay currently owns any subsequent ambient
# layering; a future discovery=False API can make this exact.
_resolve_plugin_awaitable(initialize(plugin_config))
_resolve_plugin_awaitable(relay.plugin.initialize(plugin_config))
except Exception as exc:
self._activation = None
logger.warning(
@@ -1923,18 +1908,9 @@ def _configured_plugin_inputs(
plugins_section = config.get("plugins")
dynamic_plugins: list[Any] = []
if plugins_section:
plugin_mod = getattr(relay, "plugin", None)
load_dynamic_plugins = getattr(
plugin_mod,
"load_dynamic_plugin_activation_specs",
None,
dynamic_plugins = relay.plugin.load_dynamic_plugin_activation_specs(
config_path
)
if not callable(load_dynamic_plugins):
raise RuntimeError(
"installed NeMo Relay binding does not expose "
"plugin.load_dynamic_plugin_activation_specs"
)
dynamic_plugins = load_dynamic_plugins(config_path)
plugin_config = dict(config)
plugin_config.pop("plugins", None)
return plugin_config, dynamic_plugins
@@ -1948,27 +1924,13 @@ def _configured_plugin_inputs(
return None
def _flush_relay_subscribers(relay: Any) -> None:
"""Flush Relay without blocking a 0.7 asyncio event-loop thread."""
subscribers = getattr(relay, "subscribers", None)
flush_async = getattr(subscribers, "flush_async", None)
if callable(flush_async):
_resolve_plugin_awaitable(flush_async())
return
flush = getattr(subscribers, "flush", None)
if callable(flush):
_resolve_plugin_awaitable(flush())
"""Flush Relay without blocking an asyncio event-loop thread."""
_resolve_plugin_awaitable(relay.subscribers.flush_async())
def _clear_relay_plugins(relay: Any) -> None:
"""Clear Relay plugins through the newest available binding API."""
plugin_mod = getattr(relay, "plugin", None)
clear_async = getattr(plugin_mod, "clear_async", None)
if callable(clear_async):
_resolve_plugin_awaitable(clear_async())
return
clear = getattr(plugin_mod, "clear", None)
if callable(clear):
_resolve_plugin_awaitable(clear())
"""Clear Relay plugins without blocking an asyncio event-loop thread."""
_resolve_plugin_awaitable(relay.plugin.clear_async())
def _resolve_plugin_awaitable(value: Any) -> Any:

99
tests/agent/test_relay_runtime_plugins.py
View File

@@ -31,13 +31,13 @@ class _FakeRelay:
initialize=self._initialize_plugins,
initialize_with_dynamic_plugins=self._initialize_dynamic_plugins,
load_dynamic_plugin_activation_specs=self._load_dynamic_plugin_specs,
clear=self._clear_plugins,
clear_async=self._clear_plugins_async,
)
self.scope = SimpleNamespace(
push=self._scope_push,
pop=self._scope_pop,
)
self.subscribers = SimpleNamespace(flush=self._flush)
self.subscribers = SimpleNamespace(flush_async=self._flush_async)
def get_scope_stack(self) -> None:
return None
@@ -71,8 +71,8 @@ class _FakeRelay:
self.events.append(("plugin.load_dynamic_specs", str(config_path)))
return self.dynamic_plugin_specs
def _clear_plugins(self) -> None:
self.events.append(("plugin.clear",))
async def _clear_plugins_async(self) -> None:
self.events.append(("plugin.clear_async",))
def _scope_push(self, name: str, scope_type: Any, **kwargs: Any) -> Any:
handle = ("scope", name, len(self.events))
@@ -82,32 +82,11 @@ class _FakeRelay:
def _scope_pop(self, handle: Any, **kwargs: Any) -> None:
self.events.append(("scope.pop", handle, kwargs))
def _flush(self) -> None:
self.events.append(("subscribers.flush",))
class _AsyncCleanupRelay(_FakeRelay):
"""Relay 0.7-shaped fake that rejects synchronous loop cleanup."""
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.plugin.clear_async = self._clear_plugins_async
self.subscribers.flush_async = self._flush_async
def _clear_plugins(self) -> None:
raise AssertionError("synchronous plugin.clear must not be called")
def _flush(self) -> None:
raise AssertionError("synchronous subscribers.flush must not be called")
async def _clear_plugins_async(self) -> None:
self.events.append(("plugin.clear_async",))
async def _flush_async(self) -> None:
self.events.append(("subscribers.flush_async",))
class _ConcurrentPublicationRelay(_AsyncCleanupRelay):
class _ConcurrentPublicationRelay(_FakeRelay):
def __init__(self) -> None:
super().__init__()
self.publication_finished = threading.Event()
@@ -145,7 +124,7 @@ def test_unset_config_disables_plugin_initialization(monkeypatch):
finally:
host.shutdown()
assert not any(event[0] == "subscribers.flush" for event in relay.events)
assert not any(event[0] == "subscribers.flush_async" for event in relay.events)
def test_relay_initializes_explicit_plugins_before_first_session_scope(
@@ -194,20 +173,6 @@ def test_later_host_retries_after_initialization_failure(explicit_static_config)
recovered_host.shutdown()
def test_missing_initialize_api_is_fail_open(explicit_static_config, caplog):
relay = _FakeRelay()
del relay.plugin.initialize
with caplog.at_level("WARNING"):
host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile")
try:
assert not host.managed_execution_enabled()
assert "does not expose plugin.initialize" in caplog.text
finally:
host.shutdown()
def test_missing_explicit_config_does_not_fall_back_to_discovery(
tmp_path,
monkeypatch,
@@ -263,19 +228,19 @@ def test_two_profile_hosts_initialize_once_and_clear_after_final_shutdown(
host_b.ensure_session({"session_id": "profile-b-session"})
host_a.shutdown()
assert ("plugin.clear",) not in relay.events
assert ("plugin.clear_async",) not in relay.events
host_b.shutdown()
assert relay.events[-2:] == [
("subscribers.flush",),
("plugin.clear",),
("subscribers.flush_async",),
("plugin.clear_async",),
]
assert relay.events.count(("plugin.initialize", {})) == 1
assert relay.events.count(("plugin.clear",)) == 1
assert relay.events.count(("plugin.clear_async",)) == 1
pop_index = next(
index for index, event in enumerate(relay.events) if event[0] == "scope.pop"
)
assert pop_index < relay.events.index(("plugin.clear",))
assert pop_index < relay.events.index(("plugin.clear_async",))
def test_plugin_initialization_inside_running_event_loop(explicit_static_config):
@@ -295,7 +260,7 @@ def test_plugin_initialization_inside_running_event_loop(explicit_static_config)
def test_static_plugin_cleanup_uses_async_apis_inside_running_event_loop(
explicit_static_config,
):
relay = _AsyncCleanupRelay()
relay = _FakeRelay()
async def run_lifecycle() -> None:
host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile")
@@ -389,10 +354,10 @@ manifest = "plugins/worker/relay-plugin.toml"
host_b.shutdown()
assert relay.events[-2:] == [
("subscribers.flush",),
("subscribers.flush_async",),
("plugin.activation.close",),
]
assert ("plugin.clear",) not in relay.events
assert ("plugin.clear_async",) not in relay.events
assert relay.events.count(("plugin.activation.close",)) == 1
pop_index = next(
index for index, event in enumerate(relay.events) if event[0] == "scope.pop"
@@ -431,8 +396,8 @@ manifest = "relay-plugin.toml"
finally:
host.shutdown()
assert ("subscribers.flush",) not in relay.events
assert ("plugin.clear",) not in relay.events
assert ("subscribers.flush_async",) not in relay.events
assert ("plugin.clear_async",) not in relay.events
def test_dynamic_activation_lifecycle_inside_running_event_loop(
@@ -448,7 +413,7 @@ manifest = "relay-plugin.toml"
encoding="utf-8",
)
monkeypatch.setenv(relay_runtime.RELAY_PLUGINS_CONFIG_ENV, str(config))
relay = _AsyncCleanupRelay()
relay = _FakeRelay()
relay.dynamic_plugin_specs = [{"plugin_id": "native.policy"}]
async def run_lifecycle() -> None:
@@ -479,7 +444,7 @@ manifest = "relay-plugin.toml"
encoding="utf-8",
)
monkeypatch.setenv(relay_runtime.RELAY_PLUGINS_CONFIG_ENV, str(config))
relay = _AsyncCleanupRelay()
relay = _FakeRelay()
relay.dynamic_plugin_specs = [{"plugin_id": "worker.policy"}]
async def run_lifecycle() -> None:
@@ -637,34 +602,6 @@ manifest = "relay-plugin.toml"
relay_runtime._PLUGIN_CONFIGURATION.reset_for_tests()
def test_missing_dynamic_initializer_disables_plugins(
tmp_path,
monkeypatch,
caplog,
):
config = tmp_path / "plugins.toml"
config.write_text(
"""
[[plugins.dynamic]]
manifest = "relay-plugin.toml"
""".strip(),
encoding="utf-8",
)
monkeypatch.setenv(relay_runtime.RELAY_PLUGINS_CONFIG_ENV, str(config))
relay = _FakeRelay()
relay.dynamic_plugin_specs = [{"plugin_id": "native.policy"}]
del relay.plugin.initialize_with_dynamic_plugins
with caplog.at_level("WARNING"):
host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile")
try:
assert not host.managed_execution_enabled()
assert relay.events == [("plugin.load_dynamic_specs", str(config))]
assert "require a binding" in caplog.text
finally:
host.shutdown()
def test_standard_dynamic_records_use_relay_toml_loader(
tmp_path,
monkeypatch,