From 0b7288ebcb3cbbb5b25da1948c98097dc76a41ac Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Fri, 7 Aug 2026 12:21:03 -0600 Subject: [PATCH] fix(relay): require explicit plugin configuration Signed-off-by: Bryan Bednarski --- agent/relay_runtime.py | 39 +++--- tests/agent/test_relay_runtime_plugins.py | 154 ++++++++++++++++++---- 2 files changed, 151 insertions(+), 42 deletions(-) diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index 0aa1965626..987fdf57a8 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -267,7 +267,10 @@ class _ProcessRelayPluginConfiguration: try: plugin_mod = getattr(relay, "plugin", None) - plugin_config, dynamic_plugins = _configured_plugin_inputs(relay) + 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, @@ -285,19 +288,14 @@ class _ProcessRelayPluginConfiguration: "returned no activation handle" ) self._activation = activation - except Exception: - logger.warning( - "Hermes Relay dynamic plugin activation failed; " - "continuing with configured and discovered " - "static plugins", - exc_info=True, - ) + except Exception as exc: + raise RuntimeError( + "Hermes Relay dynamic plugin activation failed" + ) from exc else: - logger.warning( + raise RuntimeError( "Hermes Relay dynamic plugins require a binding that " - "exposes plugin.initialize_with_dynamic_plugins; " - "continuing with configured and discovered static " - "plugins" + "exposes plugin.initialize_with_dynamic_plugins" ) if self._activation is None: @@ -307,8 +305,9 @@ class _ProcessRelayPluginConfiguration: "installed NeMo Relay binding does not expose " "plugin.initialize" ) - # Relay owns ambient file discovery and precedence. An explicit - # Hermes file, when configured, is supplied as the final overlay. + # 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)) except Exception as exc: self._activation = None @@ -1904,11 +1903,13 @@ def _load_nemo_relay() -> Any: return importlib.import_module("nemo_relay") -def _configured_plugin_inputs(relay: Any) -> tuple[dict[str, Any], list[Any]]: - """Load one explicit static overlay and dynamic host specs.""" +def _configured_plugin_inputs( + relay: Any, +) -> tuple[dict[str, Any], list[Any]] | None: + """Load environment-selected plugin inputs, or leave plugins disabled.""" configured = os.environ.get(RELAY_PLUGINS_CONFIG_ENV, "").strip() if not configured: - return {}, [] + return None config_path = Path(configured).expanduser() try: @@ -1943,11 +1944,11 @@ def _configured_plugin_inputs(relay: Any) -> tuple[dict[str, Any], list[Any]]: except Exception: logger.warning( "Hermes Relay plugin configuration could not be loaded from %s; " - "continuing with discovered static plugins", + "continuing without Relay plugins", config_path, exc_info=True, ) - return {}, [] + return None def _dynamic_plugin_specs( diff --git a/tests/agent/test_relay_runtime_plugins.py b/tests/agent/test_relay_runtime_plugins.py index 290f8d7658..d8cf000aac 100644 --- a/tests/agent/test_relay_runtime_plugins.py +++ b/tests/agent/test_relay_runtime_plugins.py @@ -124,7 +124,33 @@ def _reset_runtime(): relay_runtime._reset_for_tests() -def test_relay_discovers_plugins_before_first_session_scope(): +@pytest.fixture +def explicit_static_config(tmp_path, monkeypatch): + config = tmp_path / "plugins.toml" + config.write_text("", encoding="utf-8") + monkeypatch.setenv(relay_runtime.RELAY_PLUGINS_CONFIG_ENV, str(config)) + return config + + +def test_unset_config_disables_plugin_initialization(monkeypatch): + monkeypatch.delenv(relay_runtime.RELAY_PLUGINS_CONFIG_ENV, raising=False) + relay = _FakeRelay() + host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile") + + try: + assert not host.managed_execution_enabled() + host.ensure_session({"session_id": "session"}) + assert relay.events[0][0:2] == ("scope.push", relay_runtime.SESSION_SCOPE) + assert not any(event[0].startswith("plugin.") for event in relay.events) + finally: + host.shutdown() + + assert not any(event[0] == "subscribers.flush" for event in relay.events) + + +def test_relay_initializes_explicit_plugins_before_first_session_scope( + explicit_static_config, +): relay = _FakeRelay() host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile") @@ -137,7 +163,7 @@ def test_relay_discovers_plugins_before_first_session_scope(): host.shutdown() -def test_initialization_failure_is_fail_open(caplog): +def test_initialization_failure_is_fail_open(explicit_static_config, caplog): relay = _FakeRelay(initialize_error=RuntimeError("rejected config")) with caplog.at_level("WARNING"): @@ -150,7 +176,7 @@ def test_initialization_failure_is_fail_open(caplog): host.shutdown() -def test_later_host_retries_after_initialization_failure(): +def test_later_host_retries_after_initialization_failure(explicit_static_config): relay = _FakeRelay(initialize_error=RuntimeError("transient failure")) failed_host = relay_runtime.RelayRuntime(relay=relay, profile_key="failed") assert not failed_host.managed_execution_enabled() @@ -168,7 +194,7 @@ def test_later_host_retries_after_initialization_failure(): recovered_host.shutdown() -def test_missing_initialize_api_is_fail_open(caplog): +def test_missing_initialize_api_is_fail_open(explicit_static_config, caplog): relay = _FakeRelay() del relay.plugin.initialize @@ -182,7 +208,31 @@ def test_missing_initialize_api_is_fail_open(caplog): host.shutdown() -def test_two_profile_hosts_initialize_once_and_clear_after_final_shutdown(): +def test_missing_explicit_config_does_not_fall_back_to_discovery( + tmp_path, + monkeypatch, + caplog, +): + missing_config = tmp_path / "missing" / "plugins.toml" + monkeypatch.setenv( + relay_runtime.RELAY_PLUGINS_CONFIG_ENV, + str(missing_config), + ) + relay = _FakeRelay() + + with caplog.at_level("WARNING"): + host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile") + try: + assert not host.managed_execution_enabled() + assert relay.events == [] + assert "continuing without Relay plugins" in caplog.text + finally: + host.shutdown() + + +def test_two_profile_hosts_initialize_once_and_clear_after_final_shutdown( + explicit_static_config, +): relay = _FakeRelay() host_a = relay_runtime.RelayRuntime(relay=relay, profile_key="profile-a") host_b = relay_runtime.RelayRuntime(relay=relay, profile_key="profile-b") @@ -208,7 +258,7 @@ def test_two_profile_hosts_initialize_once_and_clear_after_final_shutdown(): assert pop_index < relay.events.index(("plugin.clear",)) -def test_plugin_initialization_inside_running_event_loop(): +def test_plugin_initialization_inside_running_event_loop(explicit_static_config): relay = _FakeRelay() async def construct_host() -> relay_runtime.RelayRuntime: @@ -222,7 +272,9 @@ def test_plugin_initialization_inside_running_event_loop(): host.shutdown() -def test_static_plugin_cleanup_uses_async_apis_inside_running_event_loop(): +def test_static_plugin_cleanup_uses_async_apis_inside_running_event_loop( + explicit_static_config, +): relay = _AsyncCleanupRelay() async def run_lifecycle() -> None: @@ -333,7 +385,7 @@ environment_ref = "environments/worker" assert pop_index < relay.events.index(("plugin.activation.close",)) -def test_dynamic_activation_failure_falls_back_to_discovered_static_plugins( +def test_dynamic_activation_failure_disables_plugins( tmp_path, monkeypatch, caplog, @@ -357,19 +409,16 @@ environment_ref = "environment" with caplog.at_level("WARNING"): host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile") try: - assert host.managed_execution_enabled() + assert not host.managed_execution_enabled() assert [event[0] for event in relay.events] == [ "plugin.initialize_dynamic", - "plugin.initialize", ] - assert "configured and discovered static plugins" in caplog.text + assert "dynamic plugin activation failed" in caplog.text finally: host.shutdown() - assert relay.events[-2:] == [ - ("subscribers.flush",), - ("plugin.clear",), - ] + assert ("subscribers.flush",) not in relay.events + assert ("plugin.clear",) not in relay.events def test_dynamic_activation_lifecycle_inside_running_event_loop( @@ -453,7 +502,9 @@ environment_ref = "environment" ) -def test_session_close_does_not_flush_during_concurrent_managed_publication(): +def test_session_close_does_not_flush_during_concurrent_managed_publication( + explicit_static_config, +): relay = _ConcurrentPublicationRelay() completed = threading.Event() errors: list[BaseException] = [] @@ -574,7 +625,7 @@ environment_ref = "environment" relay_runtime._PLUGIN_CONFIGURATION.reset_for_tests() -def test_missing_dynamic_initializer_falls_back_to_static_plugins( +def test_missing_dynamic_initializer_disables_plugins( tmp_path, monkeypatch, caplog, @@ -596,8 +647,8 @@ manifest_ref = "relay-plugin.toml" with caplog.at_level("WARNING"): host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile") try: - assert host.managed_execution_enabled() - assert relay.events == [("plugin.initialize", {})] + assert not host.managed_execution_enabled() + assert relay.events == [] assert "require a binding" in caplog.text finally: host.shutdown() @@ -667,9 +718,10 @@ manifest_ref = "worker/relay-plugin.toml" with caplog.at_level("WARNING"): host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile") try: - assert host.managed_execution_enabled() - assert relay.events == [("plugin.initialize", {})] + assert not host.managed_execution_enabled() + assert relay.events == [] assert "dynamic_plugins[1].plugin_id is required" in caplog.text + assert "continuing without Relay plugins" in caplog.text finally: host.shutdown() @@ -719,7 +771,61 @@ mode = "strict" ] -def test_real_binding_discovers_project_config_and_exports_native_activity( +def test_real_binding_ignores_project_config_without_explicit_opt_in( + tmp_path, + monkeypatch, +): + relay = pytest.importorskip("nemo_relay") + if getattr(relay, "_native", None) is None: + pytest.skip("NeMo Relay native binding is unavailable on this platform") + + project_root = tmp_path / "project" + working_directory = project_root / "workspace" + config_directory = project_root / ".nemo-relay" + atof_dir = tmp_path / "atof" + working_directory.mkdir(parents=True) + config_directory.mkdir() + (config_directory / "plugins.toml").write_text( + f""" +version = 1 + +[[components]] +kind = "observability" +enabled = true + +[components.config] +version = 3 + +[components.config.atof] +enabled = true + +[[components.config.atof.sinks]] +type = "file" +output_directory = "{atof_dir}" +filename = "events.jsonl" +mode = "overwrite" +""".strip(), + encoding="utf-8", + ) + xdg_config_home = tmp_path / "xdg" + xdg_config_home.mkdir() + monkeypatch.chdir(working_directory) + monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg_config_home)) + monkeypatch.delenv(relay_runtime.RELAY_PLUGINS_CONFIG_ENV, raising=False) + relay.plugin.clear() + + host = relay_runtime.RelayRuntime(relay=relay, profile_key="profile") + try: + assert not host.managed_execution_enabled() + host.ensure_session({"session_id": "native-no-plugins"}) + finally: + host.shutdown() + relay_runtime._reset_for_tests() + + assert not (atof_dir / "events.jsonl").exists() + + +def test_real_binding_loads_explicit_config_and_exports_native_activity( tmp_path, monkeypatch, ): @@ -735,7 +841,8 @@ def test_real_binding_discovers_project_config_and_exports_native_activity( atif_dir = tmp_path / "atif" working_directory.mkdir(parents=True) config_directory.mkdir() - (config_directory / "plugins.toml").write_text( + config_path = config_directory / "plugins.toml" + config_path.write_text( f""" version = 1 @@ -768,6 +875,7 @@ agent_version = "test" xdg_config_home.mkdir() monkeypatch.chdir(working_directory) monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg_config_home)) + monkeypatch.setenv(relay_runtime.RELAY_PLUGINS_CONFIG_ENV, str(config_path)) monkeypatch.setattr(relay_runtime, "_load_nemo_relay", lambda: relay) monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profile")) relay.plugin.clear()