From 1f9dedda49fecb3bc66be70350329abfdb0bff83 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 6 Jul 2026 14:20:35 -0600 Subject: [PATCH 1/9] feat(python): activate dynamic plugins Signed-off-by: Bryan Bednarski --- crates/python/src/py_plugin.rs | 129 ++++++++- .../coverage/py_plugin_coverage_tests.rs | 19 ++ python/nemo_relay/_native.pyi | 33 +++ python/nemo_relay/plugin.py | 114 +++++++- python/nemo_relay/plugin.pyi | 39 ++- python/tests/test_dynamic_plugin_host.py | 266 ++++++++++++++++++ 6 files changed, 594 insertions(+), 6 deletions(-) create mode 100644 python/tests/test_dynamic_plugin_host.py diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs index 3178471dc..28f873bab 100644 --- a/crates/python/src/py_plugin.rs +++ b/crates/python/src/py_plugin.rs @@ -31,10 +31,10 @@ use nemo_relay::api::registry::{ use nemo_relay::api::subscriber::{deregister_subscriber, register_subscriber}; use nemo_relay::error::Result as FlowResult; use nemo_relay::plugin::{ - ConfigDiagnostic, DiagnosticLevel, Plugin, PluginConfig, PluginError, PluginRegistration, - PluginRegistrationContext, active_plugin_report, clear_plugin_configuration, deregister_plugin, - initialize_plugins, list_plugin_kinds, register_plugin, rollback_registrations, - validate_plugin_config, + ConfigDiagnostic, DiagnosticLevel, DynamicPluginActivationSpec, Plugin, PluginConfig, + PluginError, PluginHostActivation, PluginRegistration, PluginRegistrationContext, + active_plugin_report, clear_plugin_configuration, deregister_plugin, initialize_plugins, + list_plugin_kinds, register_plugin, rollback_registrations, validate_plugin_config, }; use crate::convert::{json_to_py, py_to_json}; @@ -699,6 +699,112 @@ fn initialize_plugins_py<'py>( }) } +/// Owned dynamic plugin host activation. +/// +/// The public Python wrapper retains this object until ``close()`` or context +/// manager exit. Dropping it without an explicit close still clears callbacks +/// before unloading plugin code. +#[pyclass(name = "_PluginHostActivation")] +struct PyPluginHostActivation { + activation: Mutex>, + report: nemo_relay::plugin::ConfigReport, +} + +#[pymethods] +impl PyPluginHostActivation { + /// Return the activation report captured during initialization. + #[getter] + fn report(&self, py: Python<'_>) -> PyResult> { + let report = serde_json::to_value(&self.report) + .map_err(|error| pyo3::exceptions::PyRuntimeError::new_err(error.to_string()))?; + json_to_py(py, &report) + } + + /// Return whether this object still owns an active dynamic plugin host. + #[getter] + fn is_active(&self) -> PyResult { + let activation = self.activation.lock().map_err(|error| { + pyo3::exceptions::PyRuntimeError::new_err(format!( + "dynamic plugin activation lock poisoned: {error}" + )) + })?; + Ok(activation + .as_ref() + .is_some_and(PluginHostActivation::is_active)) + } + + /// Clear callbacks and unload the dynamic plugin host. + #[pyo3(signature = () -> "None", text_signature = "($self) -> None")] + fn close<'py>(&self, py: Python<'py>) -> PyResult> { + let activation = self + .activation + .lock() + .map_err(|error| { + pyo3::exceptions::PyRuntimeError::new_err(format!( + "dynamic plugin activation lock poisoned: {error}" + )) + })? + .take(); + pyo3_async_runtimes::tokio::future_into_py(py, async move { + let Some(activation) = activation else { + return Ok(()); + }; + tokio::task::spawn_blocking(move || activation.clear()) + .await + .map_err(|error| { + pyo3::exceptions::PyRuntimeError::new_err(format!( + "dynamic plugin teardown task failed: {error}" + )) + })? + .map_err(plugin_error_to_py_err) + }) + } +} + +impl Drop for PyPluginHostActivation { + fn drop(&mut self) { + let activation = self + .activation + .get_mut() + .unwrap_or_else(|error| error.into_inner()) + .take(); + if let Some(activation) = activation { + let _ = activation.clear(); + } + } +} + +/// Load and initialize an owned dynamic plugin host. +#[pyfunction(name = "activate_dynamic_plugins")] +#[pyo3(signature = (config: "object", dynamic_plugins: "object") -> "object", text_signature = "(config: object, dynamic_plugins: object) -> object")] +fn activate_dynamic_plugins_py<'py>( + py: Python<'py>, + config: &Bound<'_, PyAny>, + dynamic_plugins: &Bound<'_, PyAny>, +) -> PyResult> { + let config_json = py_to_json(config)?; + let config: PluginConfig = serde_json::from_value(config_json) + .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))?; + let dynamic_plugins_json = py_to_json(dynamic_plugins)?; + let dynamic_plugins: Vec = + serde_json::from_value(dynamic_plugins_json) + .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))?; + pyo3_async_runtimes::tokio::future_into_py(py, async move { + let (activation, report) = PluginHostActivation::activate(config, dynamic_plugins) + .await + .map_err(plugin_error_to_py_err)?; + Python::attach(|py| { + Py::new( + py, + PyPluginHostActivation { + activation: Mutex::new(Some(activation)), + report, + }, + ) + }) + }) +} + #[pyfunction(name = "clear_plugin_configuration")] #[pyo3(signature = () -> "None", text_signature = "() -> None")] fn clear_plugin_configuration_py() -> PyResult<()> { @@ -744,8 +850,10 @@ fn deregister_plugin_py(plugin_kind: &str) -> bool { pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; + m.add_class::()?; m.add_function(wrap_pyfunction!(validate_plugin_config_py, m)?)?; m.add_function(wrap_pyfunction!(initialize_plugins_py, m)?)?; + m.add_function(wrap_pyfunction!(activate_dynamic_plugins_py, m)?)?; m.add_function(wrap_pyfunction!(clear_plugin_configuration_py, m)?)?; m.add_function(wrap_pyfunction!(active_plugin_report_py, m)?)?; m.add_function(wrap_pyfunction!(list_plugin_kinds_py, m)?)?; @@ -768,6 +876,19 @@ fn to_py_err(err: impl std::fmt::Display) -> PyErr { pyo3::exceptions::PyRuntimeError::new_err(err.to_string()) } +fn plugin_error_to_py_err(error: PluginError) -> PyErr { + let message = error.to_string(); + match error { + PluginError::InvalidConfig(_) | PluginError::Serialization(_) => { + pyo3::exceptions::PyValueError::new_err(message) + } + PluginError::NotFound(_) => pyo3::exceptions::PyFileNotFoundError::new_err(message), + PluginError::Conflict(_) + | PluginError::Internal(_) + | PluginError::RegistrationFailed(_) => pyo3::exceptions::PyRuntimeError::new_err(message), + } +} + #[cfg(test)] #[path = "../tests/coverage/py_plugin_coverage_tests.rs"] mod coverage_tests; diff --git a/crates/python/tests/coverage/py_plugin_coverage_tests.rs b/crates/python/tests/coverage/py_plugin_coverage_tests.rs index 03c8557d8..522f0f9a8 100644 --- a/crates/python/tests/coverage/py_plugin_coverage_tests.rs +++ b/crates/python/tests/coverage/py_plugin_coverage_tests.rs @@ -77,8 +77,10 @@ fn register_adds_plugin_management_bindings() { for name in [ "PluginContext", + "_PluginHostActivation", "validate_plugin_config", "initialize_plugins", + "activate_dynamic_plugins", "clear_plugin_configuration", "active_plugin_report", "list_plugin_kinds", @@ -107,6 +109,23 @@ fn register_adds_plugin_management_bindings() { let report = validate_plugin_config_py(py, &config).unwrap(); let report_json = crate::convert::py_to_json(report.bind(py)).unwrap(); assert!(report_json.get("diagnostics").unwrap().is_array()); + + assert!( + plugin_error_to_py_err(PluginError::InvalidConfig("bad".into())) + .is_instance_of::(py) + ); + assert!( + plugin_error_to_py_err(PluginError::NotFound("missing".into())) + .is_instance_of::(py) + ); + assert!( + plugin_error_to_py_err(PluginError::Conflict("busy".into())) + .is_instance_of::(py) + ); + assert!( + plugin_error_to_py_err(PluginError::RegistrationFailed("teardown".into())) + .is_instance_of::(py) + ); }); } diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi index 400307416..42e022dc0 100644 --- a/python/nemo_relay/_native.pyi +++ b/python/nemo_relay/_native.pyi @@ -2249,6 +2249,39 @@ def validate_plugin_config(config: object) -> _JsonObject: """ ... +class _PluginHostActivation: + """Native owner for one process-wide dynamic plugin host.""" + + @property + def report(self) -> _JsonObject: + """Return the validation report captured during activation.""" + ... + + @property + def is_active(self) -> bool: + """Return whether this object still owns an active host.""" + ... + + def close(self) -> Awaitable[None]: + """Clear and unload this activation; repeated calls are safe.""" + ... + +def activate_dynamic_plugins(config: object, dynamic_plugins: object) -> Awaitable[_PluginHostActivation]: + """Load dynamic plugins and activate their components as one owned host. + + Args: + config: Base plugin configuration object. + dynamic_plugins: Sequence of dynamic plugin activation specifications. + + Returns: + Awaitable resolving to the native activation owner. + + Exceptional flow: + Invalid configuration, load, ownership, and registration errors are + raised through the awaitable. + """ + ... + def initialize_plugins(config: object) -> Awaitable[_JsonObject]: """Validate and activate plugin configuration. diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index 9dbb14f2e..a0e3bc2ce 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -12,7 +12,7 @@ from contextlib import asynccontextmanager from dataclasses import dataclass, field, fields, is_dataclass -from typing import TYPE_CHECKING, AsyncIterator, Callable, Literal, Protocol, TypedDict, cast +from typing import TYPE_CHECKING, AsyncIterator, Callable, Literal, Protocol, Self, TypedDict, cast from nemo_relay import ( EventSanitizeGuardrail, @@ -31,6 +31,10 @@ UnsupportedBehavior, subscribers, ) +from nemo_relay._native import _PluginHostActivation as _NativePluginHostActivation +from nemo_relay._native import ( + activate_dynamic_plugins as _activate_dynamic_plugins, +) from nemo_relay._native import ( active_plugin_report as _active_plugin_report, ) @@ -54,6 +58,8 @@ ) if TYPE_CHECKING: + from types import TracebackType + from nemo_relay import Event @@ -76,6 +82,10 @@ class ConfigReport(TypedDict): diagnostics: list[ConfigDiagnostic] +DynamicPluginKind = Literal["rust_dynamic", "worker"] +"""Execution lane for a dynamically loaded plugin.""" + + class PluginContext(Protocol): """Component-scoped registration context passed to custom plugin handlers.""" @@ -301,6 +311,80 @@ def to_dict(self) -> JsonObject: } +@dataclass(slots=True) +class DynamicPluginActivationSpec: + """One dynamic plugin component to load and activate. + + Args: + plugin_id: Expected plugin identifier from the authored manifest. + kind: Dynamic plugin execution lane. + manifest_ref: Path to the authored ``relay-plugin.toml``. + environment_ref: Optional lifecycle-managed environment path. Python + workers require the path created by Relay's plugin lifecycle. + config: Component-local JSON configuration. + """ + + plugin_id: str + kind: DynamicPluginKind + manifest_ref: str + environment_ref: str | None = None + config: JsonObject = field(default_factory=dict) + + def to_dict(self) -> JsonObject: + """Serialize this activation specification to its canonical JSON shape.""" + value: JsonObject = { + "plugin_id": self.plugin_id, + "kind": self.kind, + "manifest_ref": self.manifest_ref, + "config": _normalize_object(self.config), + } + if self.environment_ref is not None: + value["environment_ref"] = self.environment_ref + return value + + +class PluginHostActivation: + """Owned lifetime for one process-wide dynamic plugin host. + + Keep this object alive while agent code may invoke callbacks from the + loaded plugins. Prefer ``async with`` or call :meth:`close` explicitly. + Native finalization performs best-effort cleanup when an object is dropped. + """ + + __slots__ = ("_native",) + + def __init__(self, native: _NativePluginHostActivation) -> None: + self._native = native + + @property + def report(self) -> ConfigReport: + """Return the validation report captured during activation.""" + return cast(ConfigReport, self._native.report) + + @property + def is_active(self) -> bool: + """Return whether this object still owns the active plugin host.""" + return self._native.is_active + + async def close(self) -> None: + """Clear callbacks and unload plugins; repeated calls are safe.""" + await self._native.close() + + async def __aenter__(self) -> Self: + """Return this active host when entering an async context.""" + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Close the host when leaving an async context.""" + del exc_type, exc_value, traceback + await self.close() + + def validate(config: PluginConfig | JsonObject) -> ConfigReport: """Validate a plugin configuration without changing runtime state. @@ -334,6 +418,30 @@ async def initialize(config: PluginConfig | JsonObject) -> ConfigReport: return cast(ConfigReport, await _initialize_plugins(_normalize_object(config))) +async def activate_dynamic_plugins( + config: PluginConfig | JsonObject, + dynamic_plugins: list[DynamicPluginActivationSpec | JsonObject], +) -> PluginHostActivation: + """Load dynamic plugins and activate their components as one owned host. + + Args: + config: Base plugin configuration activated alongside dynamic plugins. + dynamic_plugins: Ordered activation specifications. Dataclass instances + and equivalent JSON objects may be mixed. + + Returns: + An owned activation containing the successful validation report. + + Behavior: + Only one dynamic plugin host may be active in a process. Errors roll + back partial loads. The returned object must remain alive until agent + work is complete. + """ + normalized_plugins = [_normalize_object(spec) for spec in dynamic_plugins] + native = await _activate_dynamic_plugins(_normalize_object(config), normalized_plugins) + return PluginHostActivation(native) + + def clear() -> None: """Clear the active plugin configuration. @@ -435,9 +543,13 @@ def deregister(plugin_kind: str) -> bool: "ConfigDiagnostic", "ConfigPolicy", "ConfigReport", + "DynamicPluginActivationSpec", + "DynamicPluginKind", "PluginConfig", "PluginContext", + "PluginHostActivation", "Plugin", + "activate_dynamic_plugins", "clear", "initialize", "deregister", diff --git a/python/nemo_relay/plugin.pyi b/python/nemo_relay/plugin.pyi index 3c0e5fc87..fac9db78e 100644 --- a/python/nemo_relay/plugin.pyi +++ b/python/nemo_relay/plugin.pyi @@ -2,7 +2,8 @@ # SPDX-License-Identifier: Apache-2.0 from collections.abc import Callable -from typing import AsyncContextManager, Literal, Protocol, TypedDict +from types import TracebackType +from typing import AsyncContextManager, Literal, Protocol, Self, TypedDict from nemo_relay import ( Event, @@ -21,6 +22,7 @@ from nemo_relay import ( ) UnsupportedBehavior = Literal["ignore", "warn", "error"] +DynamicPluginKind = Literal["rust_dynamic", "worker"] class _ConfigDiagnosticRequired(TypedDict): level: Literal["warning", "error"] @@ -116,8 +118,43 @@ class PluginConfig: ) -> None: ... def to_dict(self) -> JsonObject: ... +class DynamicPluginActivationSpec: + plugin_id: str + kind: DynamicPluginKind + manifest_ref: str + environment_ref: str | None + config: JsonObject + + def __init__( + self, + plugin_id: str, + kind: DynamicPluginKind, + manifest_ref: str, + environment_ref: str | None = None, + config: JsonObject = ..., + ) -> None: ... + def to_dict(self) -> JsonObject: ... + +class PluginHostActivation: + @property + def report(self) -> ConfigReport: ... + @property + def is_active(self) -> bool: ... + async def close(self) -> None: ... + async def __aenter__(self) -> Self: ... + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: ... + def validate(config: PluginConfig | JsonObject) -> ConfigReport: ... async def initialize(config: PluginConfig | JsonObject) -> ConfigReport: ... +async def activate_dynamic_plugins( + config: PluginConfig | JsonObject, + dynamic_plugins: list[DynamicPluginActivationSpec | JsonObject], +) -> PluginHostActivation: ... def clear() -> None: ... def plugin(config: PluginConfig | JsonObject) -> AsyncContextManager[ConfigReport]: ... def report() -> ConfigReport | None: ... diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py new file mode 100644 index 000000000..f2a6b30fb --- /dev/null +++ b/python/tests/test_dynamic_plugin_host.py @@ -0,0 +1,266 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""End-to-end tests for Python-owned dynamic plugin activation.""" + +from __future__ import annotations + +import gc +import hashlib +import os +import subprocess +import sys +import textwrap +import tomllib +from dataclasses import dataclass +from pathlib import Path + +import pytest + +from nemo_relay import Json, plugin, tools + + +@dataclass(frozen=True, slots=True) +class _BuiltPlugin: + plugin_id: str + kind: plugin.DynamicPluginKind + manifest: Path + + def spec(self, **config: Json) -> plugin.DynamicPluginActivationSpec: + return plugin.DynamicPluginActivationSpec( + plugin_id=self.plugin_id, + kind=self.kind, + manifest_ref=str(self.manifest), + config=config, + ) + + +def _repo_root() -> Path: + return Path(__file__).resolve().parents[2] + + +def _relay_version() -> str: + with (_repo_root() / "Cargo.toml").open("rb") as file: + return str(tomllib.load(file)["workspace"]["package"]["version"]) + + +def _native_library_name() -> str: + if sys.platform == "win32": + return "nemo_relay_plugin_fixture.dll" + if sys.platform == "darwin": + return "libnemo_relay_plugin_fixture.dylib" + return "libnemo_relay_plugin_fixture.so" + + +@pytest.fixture(scope="session") +def native_dynamic_plugin(tmp_path_factory: pytest.TempPathFactory) -> _BuiltPlugin: + root = _repo_root() + target = tmp_path_factory.mktemp("native-plugin-target") + manifest_dir = tmp_path_factory.mktemp("native-plugin-manifest") + subprocess.run( + [ + os.environ.get("CARGO", "cargo"), + "build", + "--quiet", + "--manifest-path", + str(root / "crates/core/tests/fixtures/native_plugin/Cargo.toml"), + "--target-dir", + str(target), + ], + cwd=root, + check=True, + ) + library = target / "debug" / _native_library_name() + assert library.is_file() + digest = hashlib.sha256(library.read_bytes()).hexdigest() + manifest = manifest_dir / "relay-plugin.toml" + manifest.write_text( + textwrap.dedent( + f""" + manifest_version = 1 + + [plugin] + id = "fixture_native" + kind = "rust_dynamic" + + [compat] + relay = "={_relay_version()}" + native_api = "1" + + [defaults] + enabled = false + + [capabilities] + items = ["plugin_native"] + + [integrity] + sha256 = "sha256:{digest}" + + [load] + library = {str(library)!r} + symbol = "nemo_relay_fixture_native_plugin" + """ + ) + ) + return _BuiltPlugin("fixture_native", "rust_dynamic", manifest) + + +@pytest.fixture(scope="session") +def worker_dynamic_plugin(tmp_path_factory: pytest.TempPathFactory) -> _BuiltPlugin: + root = _repo_root() + target = tmp_path_factory.mktemp("worker-plugin-target") + manifest_dir = tmp_path_factory.mktemp("worker-plugin-manifest") + subprocess.run( + [ + os.environ.get("CARGO", "cargo"), + "build", + "--quiet", + "--locked", + "--manifest-path", + str(root / "crates/core/tests/fixtures/worker_plugin/Cargo.toml"), + "--target-dir", + str(target), + ], + cwd=root, + check=True, + ) + executable = target / "debug" / ("nemo-relay-worker-plugin-fixture" + (".exe" if sys.platform == "win32" else "")) + assert executable.is_file() + manifest = manifest_dir / "relay-plugin.toml" + manifest.write_text( + textwrap.dedent( + f""" + manifest_version = 1 + + [plugin] + id = "fixture_worker" + kind = "worker" + + [compat] + relay = "={_relay_version()}" + worker_protocol = "grpc-v1" + + [defaults] + enabled = false + + [capabilities] + items = ["plugin_worker"] + + [load] + runtime = "rust" + entrypoint = {str(executable)!r} + """ + ) + ) + return _BuiltPlugin("fixture_worker", "worker", manifest) + + +def test_dynamic_plugin_activation_spec_serializes_canonical_shape(): + spec = plugin.DynamicPluginActivationSpec( + plugin_id="example.plugin", + kind="worker", + manifest_ref="/plugins/example/relay-plugin.toml", + environment_ref="/plugins/example/.venv", + config={"enabled": True}, + ) + + assert spec.to_dict() == { + "plugin_id": "example.plugin", + "kind": "worker", + "manifest_ref": "/plugins/example/relay-plugin.toml", + "environment_ref": "/plugins/example/.venv", + "config": {"enabled": True}, + } + + +async def test_native_activation_context_owns_callbacks_and_close_is_idempotent( + native_dynamic_plugin: _BuiltPlugin, +): + activation = await plugin.activate_dynamic_plugins(plugin.PluginConfig(), [native_dynamic_plugin.spec()]) + assert activation.is_active + assert activation.report == {"diagnostics": []} + + async with activation as active: + result = await tools.execute("python-native-fixture", {"input": True}, lambda args: {"args": args}) + assert active is activation + assert result["native_plugin_tool_execution"] is True + assert result["args"]["native_plugin_tool_execution_request"] is True + + assert not activation.is_active + await activation.close() + result = await tools.execute("python-native-after-close", {"input": True}, lambda args: {"args": args}) + assert "native_plugin_tool_execution" not in result + assert result == {"args": {"input": True}} + + +async def test_activation_reports_conflicts_and_rolls_back_partial_loads( + native_dynamic_plugin: _BuiltPlugin, + tmp_path: Path, +): + activation = await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + try: + with pytest.raises(RuntimeError, match="active dynamic plugin host"): + await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + with pytest.raises(RuntimeError, match="active dynamic plugin host"): + await plugin.initialize({}) + with pytest.raises(RuntimeError, match="active dynamic plugin host"): + plugin.clear() + finally: + await activation.close() + + missing = plugin.DynamicPluginActivationSpec( + plugin_id="missing_native", + kind="rust_dynamic", + manifest_ref=str(tmp_path / "missing-relay-plugin.toml"), + ) + with pytest.raises(FileNotFoundError, match="missing-relay-plugin.toml"): + await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec(), missing]) + + assert "fixture_native" not in plugin.list_kinds() + retry = await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + await retry.close() + + +async def test_invalid_dynamic_inputs_raise_normal_python_exceptions(native_dynamic_plugin: _BuiltPlugin): + with pytest.raises(ValueError, match="unknown variant"): + await plugin.activate_dynamic_plugins( + {}, + [ + { + "plugin_id": "invalid", + "kind": "invalid", + "manifest_ref": str(native_dynamic_plugin.manifest), + } + ], + ) + + with pytest.raises(ValueError, match="fixture rejection requested"): + await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec(reject=True)]) + + assert "fixture_native" not in plugin.list_kinds() + + +async def test_native_activation_finalizer_releases_callbacks(native_dynamic_plugin: _BuiltPlugin): + activation = await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + assert "fixture_native" in plugin.list_kinds() + + del activation + gc.collect() + + assert "fixture_native" not in plugin.list_kinds() + result = await tools.execute("python-native-after-finalize", {"input": True}, lambda args: args) + assert result == {"input": True} + + +async def test_worker_activation_executes_and_releases_callbacks(worker_dynamic_plugin: _BuiltPlugin): + activation = await plugin.activate_dynamic_plugins({}, [worker_dynamic_plugin.spec()]) + try: + result = await tools.execute("python-worker-fixture", {"input": True}, lambda args: {"args": args}) + assert result["worker_plugin_tool_execution"] is True + assert result["args"]["worker_plugin_tool_execution_request"] is True + finally: + await activation.close() + + assert not activation.is_active + result = await tools.execute("python-worker-after-close", {"input": True}, lambda args: {"args": args}) + assert result == {"args": {"input": True}} From ad86a5592acc6b2d06e4d7f18b4016c0d7b5266d Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 6 Jul 2026 14:57:05 -0600 Subject: [PATCH 2/9] test(python): await activation finalizer scheduling Signed-off-by: Bryan Bednarski --- python/tests/test_dynamic_plugin_host.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py index f2a6b30fb..8f77baeb6 100644 --- a/python/tests/test_dynamic_plugin_host.py +++ b/python/tests/test_dynamic_plugin_host.py @@ -5,6 +5,7 @@ from __future__ import annotations +import asyncio import gc import hashlib import os @@ -245,6 +246,9 @@ async def test_native_activation_finalizer_releases_callbacks(native_dynamic_plu assert "fixture_native" in plugin.list_kinds() del activation + # The asyncio Future returned by the native binding retains its completed + # result until the event loop processes the completion callback. + await asyncio.sleep(0) gc.collect() assert "fixture_native" not in plugin.list_kinds() From e26d68141db6f49128e59bc5e2fbcd4fd839ea72 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 6 Jul 2026 18:13:36 -0600 Subject: [PATCH 3/9] Fix Python dynamic plugin teardown lifecycle Signed-off-by: Bryan Bednarski --- crates/python/src/py_plugin.rs | 218 +++++++++++++++++++---- python/nemo_relay/plugin.py | 2 +- python/tests/test_dynamic_plugin_host.py | 143 ++++++++++++++- 3 files changed, 324 insertions(+), 39 deletions(-) diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs index 28f873bab..784d837d0 100644 --- a/crates/python/src/py_plugin.rs +++ b/crates/python/src/py_plugin.rs @@ -706,10 +706,182 @@ fn initialize_plugins_py<'py>( /// before unloading plugin code. #[pyclass(name = "_PluginHostActivation")] struct PyPluginHostActivation { - activation: Mutex>, + close_state: Arc, report: nemo_relay::plugin::ConfigReport, } +#[derive(Clone, Copy)] +enum PluginTeardownErrorKind { + Value, + NotFound, + Runtime, +} + +#[derive(Clone)] +struct PluginTeardownError { + kind: PluginTeardownErrorKind, + message: String, +} + +impl PluginTeardownError { + fn from_plugin_error(error: PluginError) -> Self { + let message = error.to_string(); + let kind = match error { + PluginError::InvalidConfig(_) | PluginError::Serialization(_) => { + PluginTeardownErrorKind::Value + } + PluginError::NotFound(_) => PluginTeardownErrorKind::NotFound, + PluginError::Conflict(_) + | PluginError::Internal(_) + | PluginError::RegistrationFailed(_) => PluginTeardownErrorKind::Runtime, + }; + Self { kind, message } + } + + fn runtime(message: impl Into) -> Self { + Self { + kind: PluginTeardownErrorKind::Runtime, + message: message.into(), + } + } + + fn to_py_err(&self) -> PyErr { + match self.kind { + PluginTeardownErrorKind::Value => { + pyo3::exceptions::PyValueError::new_err(self.message.clone()) + } + PluginTeardownErrorKind::NotFound => { + pyo3::exceptions::PyFileNotFoundError::new_err(self.message.clone()) + } + PluginTeardownErrorKind::Runtime => { + pyo3::exceptions::PyRuntimeError::new_err(self.message.clone()) + } + } + } +} + +type PluginTeardownResult = std::result::Result<(), PluginTeardownError>; + +enum PluginHostCloseStatus { + Active(Option), + Closing, + Closed, +} + +struct PluginHostCloseState { + status: Mutex, + completion: tokio::sync::watch::Sender>, +} + +impl PluginHostCloseState { + fn new(activation: PluginHostActivation) -> Self { + let (completion, _) = tokio::sync::watch::channel(None); + Self { + status: Mutex::new(PluginHostCloseStatus::Active(Some(activation))), + completion, + } + } + + fn is_active(&self) -> bool { + let status = self + .status + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + match &*status { + PluginHostCloseStatus::Active(activation) => activation + .as_ref() + .is_some_and(PluginHostActivation::is_active), + PluginHostCloseStatus::Closing | PluginHostCloseStatus::Closed => false, + } + } + + fn begin_close(self: &Arc) { + let activation = { + let mut status = self + .status + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + match &mut *status { + PluginHostCloseStatus::Active(activation) => { + let activation = activation.take(); + *status = PluginHostCloseStatus::Closing; + activation + } + PluginHostCloseStatus::Closing | PluginHostCloseStatus::Closed => None, + } + }; + let Some(activation) = activation else { + return; + }; + + // Keep the activation outside the spawned closure so a thread-spawn + // failure cannot drop it and synchronously run teardown on the caller. + let activation = Arc::new(Mutex::new(Some(activation))); + let worker_activation = Arc::clone(&activation); + let close_state = Arc::clone(self); + let spawn = std::thread::Builder::new() + .name("nemo-relay-python-plugin-teardown".into()) + .spawn(move || { + let activation = worker_activation + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); + let result = match activation { + Some(activation) => { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + activation.clear() + })) + .map_err(|_| { + PluginTeardownError::runtime("dynamic plugin teardown task panicked") + }) + .and_then(|result| result.map_err(PluginTeardownError::from_plugin_error)) + } + None => Err(PluginTeardownError::runtime( + "dynamic plugin teardown task lost its activation", + )), + }; + close_state.finish(result); + }); + + if let Err(error) = spawn { + // Cleanup must never fall back to the Python thread. Retain the + // activation for process lifetime if no teardown thread can start. + if let Some(activation) = activation + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + { + std::mem::forget(activation); + } + self.finish(Err(PluginTeardownError::runtime(format!( + "failed to start dynamic plugin teardown task: {error}" + )))); + } + } + + fn finish(&self, result: PluginTeardownResult) { + *self + .status + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = PluginHostCloseStatus::Closed; + self.completion.send_replace(Some(result)); + } + + async fn wait_for_close(&self) -> PluginTeardownResult { + let mut completion = self.completion.subscribe(); + loop { + if let Some(result) = completion.borrow().clone() { + return result; + } + if completion.changed().await.is_err() { + return Err(PluginTeardownError::runtime( + "dynamic plugin teardown result channel closed unexpectedly", + )); + } + } + } +} + #[pymethods] impl PyPluginHostActivation { /// Return the activation report captured during initialization. @@ -723,54 +895,26 @@ impl PyPluginHostActivation { /// Return whether this object still owns an active dynamic plugin host. #[getter] fn is_active(&self) -> PyResult { - let activation = self.activation.lock().map_err(|error| { - pyo3::exceptions::PyRuntimeError::new_err(format!( - "dynamic plugin activation lock poisoned: {error}" - )) - })?; - Ok(activation - .as_ref() - .is_some_and(PluginHostActivation::is_active)) + Ok(self.close_state.is_active()) } /// Clear callbacks and unload the dynamic plugin host. #[pyo3(signature = () -> "None", text_signature = "($self) -> None")] fn close<'py>(&self, py: Python<'py>) -> PyResult> { - let activation = self - .activation - .lock() - .map_err(|error| { - pyo3::exceptions::PyRuntimeError::new_err(format!( - "dynamic plugin activation lock poisoned: {error}" - )) - })? - .take(); + let close_state = Arc::clone(&self.close_state); + close_state.begin_close(); pyo3_async_runtimes::tokio::future_into_py(py, async move { - let Some(activation) = activation else { - return Ok(()); - }; - tokio::task::spawn_blocking(move || activation.clear()) + close_state + .wait_for_close() .await - .map_err(|error| { - pyo3::exceptions::PyRuntimeError::new_err(format!( - "dynamic plugin teardown task failed: {error}" - )) - })? - .map_err(plugin_error_to_py_err) + .map_err(|error| error.to_py_err()) }) } } impl Drop for PyPluginHostActivation { fn drop(&mut self) { - let activation = self - .activation - .get_mut() - .unwrap_or_else(|error| error.into_inner()) - .take(); - if let Some(activation) = activation { - let _ = activation.clear(); - } + self.close_state.begin_close(); } } @@ -797,7 +941,7 @@ fn activate_dynamic_plugins_py<'py>( Py::new( py, PyPluginHostActivation { - activation: Mutex::new(Some(activation)), + close_state: Arc::new(PluginHostCloseState::new(activation)), report, }, ) diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index a0e3bc2ce..a1b216904 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -221,7 +221,7 @@ def _normalize(value: object) -> Json: if isinstance(value, list): return [_normalize(item) for item in value] if isinstance(value, dict): - return {cast(str, key): _normalize(val) for key, val in value.items() if val is not None} + return {cast(str, key): _normalize(val) for key, val in value.items()} return cast(Json, value) diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py index 8f77baeb6..061d23320 100644 --- a/python/tests/test_dynamic_plugin_host.py +++ b/python/tests/test_dynamic_plugin_host.py @@ -9,16 +9,20 @@ import gc import hashlib import os +import signal import subprocess import sys import textwrap +import threading +import time import tomllib from dataclasses import dataclass from pathlib import Path +from typing import cast import pytest -from nemo_relay import Json, plugin, tools +from nemo_relay import Json, plugin, scope, tools @dataclass(frozen=True, slots=True) @@ -174,6 +178,25 @@ def test_dynamic_plugin_activation_spec_serializes_canonical_shape(): } +def test_dynamic_plugin_activation_spec_preserves_nested_json_nulls(): + spec = plugin.DynamicPluginActivationSpec( + plugin_id="example.plugin", + kind="worker", + manifest_ref="/plugins/example/relay-plugin.toml", + config={ + "top_level": None, + "nested": {"value": None}, + "items": [None, {"value": None}], + }, + ) + + assert spec.to_dict()["config"] == { + "top_level": None, + "nested": {"value": None}, + "items": [None, {"value": None}], + } + + async def test_native_activation_context_owns_callbacks_and_close_is_idempotent( native_dynamic_plugin: _BuiltPlugin, ): @@ -194,6 +217,60 @@ async def test_native_activation_context_owns_callbacks_and_close_is_idempotent( assert result == {"args": {"input": True}} +async def test_concurrent_close_waiters_share_cancellation_resistant_teardown( + native_dynamic_plugin: _BuiltPlugin, +): + started = threading.Event() + release = threading.Event() + plugin_kind = "python.dynamic_close_waiter" + + class BlockingSubscriberPlugin: + def validate(self, _plugin_config): + return None + + def register(self, _plugin_config, context): + def block(_event): + started.set() + assert release.wait(timeout=5) + + context.register_subscriber("block_teardown", block) + + plugin.register(plugin_kind, cast(plugin.Plugin, BlockingSubscriberPlugin())) + activation = await plugin.activate_dynamic_plugins( + { + "version": 1, + "components": [{"kind": plugin_kind, "config": {}}], + }, + [native_dynamic_plugin.spec()], + ) + second_close: asyncio.Task[None] | None = None + try: + scope.event("python-dynamic-close-blocker") + assert await asyncio.to_thread(started.wait, 2) + + first_close = asyncio.create_task(activation.close()) + while activation.is_active: + await asyncio.sleep(0) + first_close.cancel() + with pytest.raises(asyncio.CancelledError): + await first_close + + second_close = asyncio.create_task(activation.close()) + await asyncio.sleep(0.05) + assert not second_close.done() + + release.set() + await second_close + await activation.close() + assert not activation.is_active + finally: + release.set() + if second_close is not None: + await asyncio.gather(second_close, return_exceptions=True) + await activation.close() + plugin.deregister(plugin_kind) + + async def test_activation_reports_conflicts_and_rolls_back_partial_loads( native_dynamic_plugin: _BuiltPlugin, tmp_path: Path, @@ -251,11 +328,75 @@ async def test_native_activation_finalizer_releases_callbacks(native_dynamic_plu await asyncio.sleep(0) gc.collect() + for _ in range(100): + if "fixture_native" not in plugin.list_kinds(): + break + await asyncio.sleep(0.01) assert "fixture_native" not in plugin.list_kinds() result = await tools.execute("python-native-after-finalize", {"input": True}, lambda args: args) assert result == {"input": True} +@pytest.mark.skipif(os.name == "nt", reason="requires POSIX worker stop/continue signals") +async def test_worker_activation_finalizer_never_waits_on_python_thread( + worker_dynamic_plugin: _BuiltPlugin, + tmp_path: Path, +): + with worker_dynamic_plugin.manifest.open("rb") as file: + worker_entrypoint = Path(tomllib.load(file)["load"]["entrypoint"]) + + pid_file = tmp_path / "worker.pid" + wrapper = tmp_path / "worker-wrapper.sh" + wrapper.write_text(f"#!/bin/sh\nprintf '%s' \"$$\" > {str(pid_file)!r}\nexec {str(worker_entrypoint)!r}\n") + wrapper.chmod(0o755) + manifest = tmp_path / "relay-plugin.toml" + manifest.write_text( + worker_dynamic_plugin.manifest.read_text().replace( + f"entrypoint = {str(worker_entrypoint)!r}", + f"entrypoint = {str(wrapper)!r}", + ) + ) + + activation = await plugin.activate_dynamic_plugins( + {}, + [_BuiltPlugin("fixture_worker", "worker", manifest).spec()], + ) + native_activation = getattr(activation, "_native") + del activation + await asyncio.sleep(0) + gc.collect() + + worker_pid = int(pid_file.read_text()) + os.kill(worker_pid, signal.SIGSTOP) + resumer = subprocess.Popen( + [ + "/bin/sh", + "-c", + 'sleep 0.8; kill -CONT "$1"', + "resume-worker", + str(worker_pid), + ] + ) + started_at = time.perf_counter() + try: + del native_activation + gc.collect() + elapsed = time.perf_counter() - started_at + finally: + try: + os.kill(worker_pid, signal.SIGCONT) + except ProcessLookupError: + pass + resumer.wait(timeout=5) + + assert elapsed < 0.4 + for _ in range(500): + if "fixture_worker" not in plugin.list_kinds(): + break + await asyncio.sleep(0.01) + assert "fixture_worker" not in plugin.list_kinds() + + async def test_worker_activation_executes_and_releases_callbacks(worker_dynamic_plugin: _BuiltPlugin): activation = await plugin.activate_dynamic_plugins({}, [worker_dynamic_plugin.spec()]) try: From 566e18f8c430d069063d8e8664114862fb598318 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 7 Jul 2026 13:47:47 -0600 Subject: [PATCH 4/9] fix(python): preserve static-only plugin activation Signed-off-by: Bryan Bednarski --- python/nemo_relay/plugin.py | 10 ++++++---- python/tests/test_dynamic_plugin_host.py | 10 ++++++++++ 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index a1b216904..71f72368d 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -425,9 +425,10 @@ async def activate_dynamic_plugins( """Load dynamic plugins and activate their components as one owned host. Args: - config: Base plugin configuration activated alongside dynamic plugins. - dynamic_plugins: Ordered activation specifications. Dataclass instances - and equivalent JSON objects may be mixed. + config: Base plugin configuration activated before the dynamic + components. It may contain statically registered components. + dynamic_plugins: Non-empty ordered activation specifications. Dataclass + instances and equivalent JSON objects may be mixed. Returns: An owned activation containing the successful validation report. @@ -435,7 +436,8 @@ async def activate_dynamic_plugins( Behavior: Only one dynamic plugin host may be active in a process. Errors roll back partial loads. The returned object must remain alive until agent - work is complete. + work is complete. Use :func:`initialize` when no dynamic plugins are + configured. """ normalized_plugins = [_normalize_object(spec) for spec in dynamic_plugins] native = await _activate_dynamic_plugins(_normalize_object(config), normalized_plugins) diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py index 061d23320..800cd6c4c 100644 --- a/python/tests/test_dynamic_plugin_host.py +++ b/python/tests/test_dynamic_plugin_host.py @@ -197,6 +197,16 @@ def test_dynamic_plugin_activation_spec_preserves_nested_json_nulls(): } +async def test_empty_dynamic_specs_preserve_static_initialization_path(): + with pytest.raises(ValueError, match="at least one dynamic plugin"): + await plugin.activate_dynamic_plugins(plugin.PluginConfig(), []) + + assert plugin.report() is None + report = await plugin.initialize(plugin.PluginConfig()) + assert report == {"diagnostics": []} + plugin.clear() + + async def test_native_activation_context_owns_callbacks_and_close_is_idempotent( native_dynamic_plugin: _BuiltPlugin, ): From 66434474a93b0f9017a613c372147b169bb25f36 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 13 Jul 2026 21:06:03 -0600 Subject: [PATCH 5/9] docs(python): clarify plugin activation state Signed-off-by: Bryan Bednarski --- crates/python/src/py_plugin.rs | 5 ++++- python/nemo_relay/_native.pyi | 6 +++++- python/nemo_relay/plugin.py | 6 +++++- 3 files changed, 14 insertions(+), 3 deletions(-) diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs index 784d837d0..b5ef5e492 100644 --- a/crates/python/src/py_plugin.rs +++ b/crates/python/src/py_plugin.rs @@ -892,7 +892,10 @@ impl PyPluginHostActivation { json_to_py(py, &report) } - /// Return whether this object still owns an active dynamic plugin host. + /// Return whether this activation handle has not begun teardown. + /// + /// `False` does not guarantee another process-wide activation can start; + /// failed teardown may intentionally retain the activation owner. #[getter] fn is_active(&self) -> PyResult { Ok(self.close_state.is_active()) diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi index 42e022dc0..98e56b189 100644 --- a/python/nemo_relay/_native.pyi +++ b/python/nemo_relay/_native.pyi @@ -2259,7 +2259,11 @@ class _PluginHostActivation: @property def is_active(self) -> bool: - """Return whether this object still owns an active host.""" + """Return whether this activation handle has not begun teardown. + + ``False`` does not guarantee another process-wide activation can start; + failed teardown may intentionally retain the activation owner. + """ ... def close(self) -> Awaitable[None]: diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index 71f72368d..f62dbee6f 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -363,7 +363,11 @@ def report(self) -> ConfigReport: @property def is_active(self) -> bool: - """Return whether this object still owns the active plugin host.""" + """Return whether this activation handle has not begun teardown. + + ``False`` does not guarantee another process-wide activation can start; + failed teardown may intentionally retain the activation owner. + """ return self._native.is_active async def close(self) -> None: From 2092ded1c6f8a08d76d51a441b119679c5dffd79 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 13 Jul 2026 23:29:36 -0600 Subject: [PATCH 6/9] fix(python): layer discovered plugin configuration Signed-off-by: Bryan Bednarski --- crates/python/src/py_plugin.rs | 7 ++-- python/nemo_relay/plugin.py | 5 ++- python/tests/test_dynamic_plugin_host.py | 50 ++++++++++++++++++++++++ 3 files changed, 57 insertions(+), 5 deletions(-) diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs index b5ef5e492..19dee9d63 100644 --- a/crates/python/src/py_plugin.rs +++ b/crates/python/src/py_plugin.rs @@ -937,9 +937,10 @@ fn activate_dynamic_plugins_py<'py>( serde_json::from_value(dynamic_plugins_json) .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))?; pyo3_async_runtimes::tokio::future_into_py(py, async move { - let (activation, report) = PluginHostActivation::activate(config, dynamic_plugins) - .await - .map_err(plugin_error_to_py_err)?; + let (activation, report) = + PluginHostActivation::activate_with_discovered_config(config, dynamic_plugins) + .await + .map_err(plugin_error_to_py_err)?; Python::attach(|py| { Py::new( py, diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index f62dbee6f..b0c55e025 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -429,8 +429,9 @@ async def activate_dynamic_plugins( """Load dynamic plugins and activate their components as one owned host. Args: - config: Base plugin configuration activated before the dynamic - components. It may contain statically registered components. + config: Base plugin configuration layered over the discovered + ``plugins.toml`` files before dynamic components are appended. It + may contain statically registered components. dynamic_plugins: Non-empty ordered activation specifications. Dataclass instances and equivalent JSON objects may be mixed. diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py index 800cd6c4c..6ff051a5b 100644 --- a/python/tests/test_dynamic_plugin_host.py +++ b/python/tests/test_dynamic_plugin_host.py @@ -227,6 +227,56 @@ async def test_native_activation_context_owns_callbacks_and_close_is_idempotent( assert result == {"args": {"input": True}} +async def test_dynamic_activation_layers_plugins_toml_static_components( + native_dynamic_plugin: _BuiltPlugin, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + static_kind = "python.fixture.file-static-base" + + class FileStaticPlugin: + def validate(self, _plugin_config): + return None + + def register(self, _plugin_config, context): + context.register_tool_request_intercept( + "mark-file-static-base", + 0, + False, + lambda _name, args: {**args, "file_static_base": True}, + ) + + project_config = tmp_path / ".nemo-relay" + project_config.mkdir() + (project_config / "plugins.toml").write_text( + textwrap.dedent( + f""" + version = 1 + + [[components]] + kind = {static_kind!r} + enabled = true + """ + ) + ) + isolated_user_config = tmp_path / "xdg" + isolated_user_config.mkdir() + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("XDG_CONFIG_HOME", str(isolated_user_config)) + + plugin.register(static_kind, cast(plugin.Plugin, FileStaticPlugin())) + activation = None + try: + activation = await plugin.activate_dynamic_plugins(plugin.PluginConfig(), [native_dynamic_plugin.spec()]) + result = await tools.execute("python-file-static-base", {"input": True}, lambda args: args) + assert result["file_static_base"] is True + assert result["native_plugin_tool_execution"] is True + finally: + if activation is not None: + await activation.close() + plugin.deregister(static_kind) + + async def test_concurrent_close_waiters_share_cancellation_resistant_teardown( native_dynamic_plugin: _BuiltPlugin, ): From bfd9b305d78dca6ff6557f979855c3e18ae0e623 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 02:15:19 -0600 Subject: [PATCH 7/9] fix(python): preserve plugin config null semantics Signed-off-by: Bryan Bednarski --- python/nemo_relay/plugin.py | 23 +++++--- python/tests/test_dynamic_plugin_host.py | 75 ++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 7 deletions(-) diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index b0c55e025..161cff8b8 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -209,19 +209,24 @@ class _SupportsToDict(Protocol): def to_dict(self) -> JsonObject: ... -def _normalize(value: object) -> Json: +def _normalize(value: object, *, preserve_nulls: bool = False) -> Json: if hasattr(value, "to_dict"): return cast(_SupportsToDict, value).to_dict() if is_dataclass(value) and not isinstance(value, type): return { - field_info.name: _normalize(field_value) + field_info.name: _normalize(field_value, preserve_nulls=preserve_nulls) for field_info in fields(value) - if (field_value := getattr(value, field_info.name)) is not None + for field_value in [getattr(value, field_info.name)] + if preserve_nulls or field_value is not None } if isinstance(value, list): - return [_normalize(item) for item in value] + return [_normalize(item, preserve_nulls=preserve_nulls) for item in value] if isinstance(value, dict): - return {cast(str, key): _normalize(val) for key, val in value.items()} + return { + cast(str, key): _normalize(val, preserve_nulls=preserve_nulls or key == "config") + for key, val in value.items() + if preserve_nulls or val is not None + } return cast(Json, value) @@ -229,6 +234,10 @@ def _normalize_object(value: object) -> JsonObject: return cast(JsonObject, _normalize(value)) +def _normalize_component_config(value: object) -> JsonObject: + return cast(JsonObject, _normalize(value, preserve_nulls=True)) + + @dataclass(slots=True) class ConfigPolicy: """Policy for unsupported plugin configuration. @@ -280,7 +289,7 @@ def to_dict(self) -> JsonObject: return { "kind": self.kind, "enabled": self.enabled, - "config": _normalize_object(self.config), + "config": _normalize_component_config(self.config), } @@ -336,7 +345,7 @@ def to_dict(self) -> JsonObject: "plugin_id": self.plugin_id, "kind": self.kind, "manifest_ref": self.manifest_ref, - "config": _normalize_object(self.config), + "config": _normalize_component_config(self.config), } if self.environment_ref is not None: value["environment_ref"] = self.environment_ref diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py index 6ff051a5b..173aa58fa 100644 --- a/python/tests/test_dynamic_plugin_host.py +++ b/python/tests/test_dynamic_plugin_host.py @@ -197,6 +197,81 @@ def test_dynamic_plugin_activation_spec_preserves_nested_json_nulls(): } +def test_validate_omits_raw_plugin_config_nulls_but_preserves_component_config_nulls( + monkeypatch: pytest.MonkeyPatch, +): + captured: list[object] = [] + + def validate(config: object) -> object: + captured.append(config) + return {"diagnostics": []} + + monkeypatch.setattr(plugin, "_validate_plugin_config", validate) + + assert plugin.validate( + { + "version": None, + "policy": {"unknown_component": None}, + "components": [ + { + "kind": "example", + "enabled": None, + "config": {"top_level": None, "nested": {"value": None}}, + } + ], + } + ) == {"diagnostics": []} + assert captured == [ + { + "policy": {}, + "components": [ + { + "kind": "example", + "config": {"top_level": None, "nested": {"value": None}}, + } + ], + } + ] + + +def test_validate_raw_plugin_config_nulls_as_omitted_fields(): + assert plugin.validate({"version": None, "components": None, "policy": None}) == {"diagnostics": []} + + +async def test_initialize_omits_raw_plugin_config_nulls_but_preserves_component_config_nulls( + monkeypatch: pytest.MonkeyPatch, +): + captured: list[object] = [] + + async def initialize(config: object) -> object: + captured.append(config) + return {"diagnostics": []} + + monkeypatch.setattr(plugin, "_initialize_plugins", initialize) + + assert await plugin.initialize( + { + "version": None, + "components": [ + { + "kind": "example", + "config": {"top_level": None, "items": [None, {"value": None}]}, + } + ], + } + ) == {"diagnostics": []} + assert captured == [ + { + "components": [ + { + "kind": "example", + "config": {"top_level": None, "items": [None, {"value": None}]}, + } + ], + } + ] + + async def test_empty_dynamic_specs_preserve_static_initialization_path(): with pytest.raises(ValueError, match="at least one dynamic plugin"): await plugin.activate_dynamic_plugins(plugin.PluginConfig(), []) From ae60c7cc27df26b056d8a1ca1b1fc19473eff036 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 07:22:28 -0600 Subject: [PATCH 8/9] fix(python): harden dynamic plugin activation Signed-off-by: Bryan Bednarski --- crates/python/src/py_plugin.rs | 11 +---------- python/tests/test_dynamic_plugin_host.py | 4 ++-- 2 files changed, 3 insertions(+), 12 deletions(-) diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs index 19dee9d63..73ef872fb 100644 --- a/crates/python/src/py_plugin.rs +++ b/crates/python/src/py_plugin.rs @@ -1025,16 +1025,7 @@ fn to_py_err(err: impl std::fmt::Display) -> PyErr { } fn plugin_error_to_py_err(error: PluginError) -> PyErr { - let message = error.to_string(); - match error { - PluginError::InvalidConfig(_) | PluginError::Serialization(_) => { - pyo3::exceptions::PyValueError::new_err(message) - } - PluginError::NotFound(_) => pyo3::exceptions::PyFileNotFoundError::new_err(message), - PluginError::Conflict(_) - | PluginError::Internal(_) - | PluginError::RegistrationFailed(_) => pyo3::exceptions::PyRuntimeError::new_err(message), - } + PluginTeardownError::from_plugin_error(error).to_py_err() } #[cfg(test)] diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py index 173aa58fa..19464721b 100644 --- a/python/tests/test_dynamic_plugin_host.py +++ b/python/tests/test_dynamic_plugin_host.py @@ -102,7 +102,7 @@ def native_dynamic_plugin(tmp_path_factory: pytest.TempPathFactory) -> _BuiltPlu sha256 = "sha256:{digest}" [load] - library = {str(library)!r} + library = {library.as_posix()!r} symbol = "nemo_relay_fixture_native_plugin" """ ) @@ -153,7 +153,7 @@ def worker_dynamic_plugin(tmp_path_factory: pytest.TempPathFactory) -> _BuiltPlu [load] runtime = "rust" - entrypoint = {str(executable)!r} + entrypoint = {executable.as_posix()!r} """ ) ) From f6954c364c08ec40f4b9e6e3e20523d987ed9d71 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 09:14:13 -0600 Subject: [PATCH 9/9] refactor(python): clarify dynamic plugin initialization Signed-off-by: Bryan Bednarski --- crates/python/src/py_plugin.rs | 8 +++--- .../coverage/py_plugin_coverage_tests.rs | 2 +- python/nemo_relay/_native.pyi | 4 +-- python/nemo_relay/plugin.py | 18 ++++++------- python/nemo_relay/plugin.pyi | 2 +- python/tests/test_dynamic_plugin_host.py | 26 +++++++++---------- 6 files changed, 30 insertions(+), 30 deletions(-) diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs index 73ef872fb..fdf3455a1 100644 --- a/crates/python/src/py_plugin.rs +++ b/crates/python/src/py_plugin.rs @@ -921,10 +921,10 @@ impl Drop for PyPluginHostActivation { } } -/// Load and initialize an owned dynamic plugin host. -#[pyfunction(name = "activate_dynamic_plugins")] +/// Initialize registered components with an owned dynamic plugin host. +#[pyfunction(name = "initialize_with_dynamic_plugins")] #[pyo3(signature = (config: "object", dynamic_plugins: "object") -> "object", text_signature = "(config: object, dynamic_plugins: object) -> object")] -fn activate_dynamic_plugins_py<'py>( +fn initialize_with_dynamic_plugins_py<'py>( py: Python<'py>, config: &Bound<'_, PyAny>, dynamic_plugins: &Bound<'_, PyAny>, @@ -1001,7 +1001,7 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_function(wrap_pyfunction!(validate_plugin_config_py, m)?)?; m.add_function(wrap_pyfunction!(initialize_plugins_py, m)?)?; - m.add_function(wrap_pyfunction!(activate_dynamic_plugins_py, m)?)?; + m.add_function(wrap_pyfunction!(initialize_with_dynamic_plugins_py, m)?)?; m.add_function(wrap_pyfunction!(clear_plugin_configuration_py, m)?)?; m.add_function(wrap_pyfunction!(active_plugin_report_py, m)?)?; m.add_function(wrap_pyfunction!(list_plugin_kinds_py, m)?)?; diff --git a/crates/python/tests/coverage/py_plugin_coverage_tests.rs b/crates/python/tests/coverage/py_plugin_coverage_tests.rs index 522f0f9a8..7cd32027e 100644 --- a/crates/python/tests/coverage/py_plugin_coverage_tests.rs +++ b/crates/python/tests/coverage/py_plugin_coverage_tests.rs @@ -80,7 +80,7 @@ fn register_adds_plugin_management_bindings() { "_PluginHostActivation", "validate_plugin_config", "initialize_plugins", - "activate_dynamic_plugins", + "initialize_with_dynamic_plugins", "clear_plugin_configuration", "active_plugin_report", "list_plugin_kinds", diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi index 98e56b189..cbd8286fc 100644 --- a/python/nemo_relay/_native.pyi +++ b/python/nemo_relay/_native.pyi @@ -2270,8 +2270,8 @@ class _PluginHostActivation: """Clear and unload this activation; repeated calls are safe.""" ... -def activate_dynamic_plugins(config: object, dynamic_plugins: object) -> Awaitable[_PluginHostActivation]: - """Load dynamic plugins and activate their components as one owned host. +def initialize_with_dynamic_plugins(config: object, dynamic_plugins: object) -> Awaitable[_PluginHostActivation]: + """Initialize registered components with dynamic plugins as one owned host. Args: config: Base plugin configuration object. diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index 161cff8b8..c143f29c6 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -32,9 +32,6 @@ subscribers, ) from nemo_relay._native import _PluginHostActivation as _NativePluginHostActivation -from nemo_relay._native import ( - activate_dynamic_plugins as _activate_dynamic_plugins, -) from nemo_relay._native import ( active_plugin_report as _active_plugin_report, ) @@ -47,6 +44,9 @@ from nemo_relay._native import ( initialize_plugins as _initialize_plugins, ) +from nemo_relay._native import ( + initialize_with_dynamic_plugins as _initialize_with_dynamic_plugins, +) from nemo_relay._native import ( list_plugin_kinds as _list_plugin_kinds, ) @@ -431,11 +431,11 @@ async def initialize(config: PluginConfig | JsonObject) -> ConfigReport: return cast(ConfigReport, await _initialize_plugins(_normalize_object(config))) -async def activate_dynamic_plugins( +async def initialize_with_dynamic_plugins( config: PluginConfig | JsonObject, dynamic_plugins: list[DynamicPluginActivationSpec | JsonObject], ) -> PluginHostActivation: - """Load dynamic plugins and activate their components as one owned host. + """Initialize registered components with dynamic plugins as one owned host. Args: config: Base plugin configuration layered over the discovered @@ -450,11 +450,11 @@ async def activate_dynamic_plugins( Behavior: Only one dynamic plugin host may be active in a process. Errors roll back partial loads. The returned object must remain alive until agent - work is complete. Use :func:`initialize` when no dynamic plugins are - configured. + work is complete. This is the owned initialization path when dynamic + plugins are configured; use :func:`initialize` for static-only config. """ normalized_plugins = [_normalize_object(spec) for spec in dynamic_plugins] - native = await _activate_dynamic_plugins(_normalize_object(config), normalized_plugins) + native = await _initialize_with_dynamic_plugins(_normalize_object(config), normalized_plugins) return PluginHostActivation(native) @@ -565,7 +565,7 @@ def deregister(plugin_kind: str) -> bool: "PluginContext", "PluginHostActivation", "Plugin", - "activate_dynamic_plugins", + "initialize_with_dynamic_plugins", "clear", "initialize", "deregister", diff --git a/python/nemo_relay/plugin.pyi b/python/nemo_relay/plugin.pyi index fac9db78e..d5d387ff5 100644 --- a/python/nemo_relay/plugin.pyi +++ b/python/nemo_relay/plugin.pyi @@ -151,7 +151,7 @@ class PluginHostActivation: def validate(config: PluginConfig | JsonObject) -> ConfigReport: ... async def initialize(config: PluginConfig | JsonObject) -> ConfigReport: ... -async def activate_dynamic_plugins( +async def initialize_with_dynamic_plugins( config: PluginConfig | JsonObject, dynamic_plugins: list[DynamicPluginActivationSpec | JsonObject], ) -> PluginHostActivation: ... diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py index 19464721b..e0d6539e2 100644 --- a/python/tests/test_dynamic_plugin_host.py +++ b/python/tests/test_dynamic_plugin_host.py @@ -274,7 +274,7 @@ async def initialize(config: object) -> object: async def test_empty_dynamic_specs_preserve_static_initialization_path(): with pytest.raises(ValueError, match="at least one dynamic plugin"): - await plugin.activate_dynamic_plugins(plugin.PluginConfig(), []) + await plugin.initialize_with_dynamic_plugins(plugin.PluginConfig(), []) assert plugin.report() is None report = await plugin.initialize(plugin.PluginConfig()) @@ -285,7 +285,7 @@ async def test_empty_dynamic_specs_preserve_static_initialization_path(): async def test_native_activation_context_owns_callbacks_and_close_is_idempotent( native_dynamic_plugin: _BuiltPlugin, ): - activation = await plugin.activate_dynamic_plugins(plugin.PluginConfig(), [native_dynamic_plugin.spec()]) + activation = await plugin.initialize_with_dynamic_plugins(plugin.PluginConfig(), [native_dynamic_plugin.spec()]) assert activation.is_active assert activation.report == {"diagnostics": []} @@ -342,7 +342,7 @@ def register(self, _plugin_config, context): plugin.register(static_kind, cast(plugin.Plugin, FileStaticPlugin())) activation = None try: - activation = await plugin.activate_dynamic_plugins(plugin.PluginConfig(), [native_dynamic_plugin.spec()]) + activation = await plugin.initialize_with_dynamic_plugins(plugin.PluginConfig(), [native_dynamic_plugin.spec()]) result = await tools.execute("python-file-static-base", {"input": True}, lambda args: args) assert result["file_static_base"] is True assert result["native_plugin_tool_execution"] is True @@ -371,7 +371,7 @@ def block(_event): context.register_subscriber("block_teardown", block) plugin.register(plugin_kind, cast(plugin.Plugin, BlockingSubscriberPlugin())) - activation = await plugin.activate_dynamic_plugins( + activation = await plugin.initialize_with_dynamic_plugins( { "version": 1, "components": [{"kind": plugin_kind, "config": {}}], @@ -410,10 +410,10 @@ async def test_activation_reports_conflicts_and_rolls_back_partial_loads( native_dynamic_plugin: _BuiltPlugin, tmp_path: Path, ): - activation = await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + activation = await plugin.initialize_with_dynamic_plugins({}, [native_dynamic_plugin.spec()]) try: with pytest.raises(RuntimeError, match="active dynamic plugin host"): - await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + await plugin.initialize_with_dynamic_plugins({}, [native_dynamic_plugin.spec()]) with pytest.raises(RuntimeError, match="active dynamic plugin host"): await plugin.initialize({}) with pytest.raises(RuntimeError, match="active dynamic plugin host"): @@ -427,16 +427,16 @@ async def test_activation_reports_conflicts_and_rolls_back_partial_loads( manifest_ref=str(tmp_path / "missing-relay-plugin.toml"), ) with pytest.raises(FileNotFoundError, match="missing-relay-plugin.toml"): - await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec(), missing]) + await plugin.initialize_with_dynamic_plugins({}, [native_dynamic_plugin.spec(), missing]) assert "fixture_native" not in plugin.list_kinds() - retry = await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + retry = await plugin.initialize_with_dynamic_plugins({}, [native_dynamic_plugin.spec()]) await retry.close() async def test_invalid_dynamic_inputs_raise_normal_python_exceptions(native_dynamic_plugin: _BuiltPlugin): with pytest.raises(ValueError, match="unknown variant"): - await plugin.activate_dynamic_plugins( + await plugin.initialize_with_dynamic_plugins( {}, [ { @@ -448,13 +448,13 @@ async def test_invalid_dynamic_inputs_raise_normal_python_exceptions(native_dyna ) with pytest.raises(ValueError, match="fixture rejection requested"): - await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec(reject=True)]) + await plugin.initialize_with_dynamic_plugins({}, [native_dynamic_plugin.spec(reject=True)]) assert "fixture_native" not in plugin.list_kinds() async def test_native_activation_finalizer_releases_callbacks(native_dynamic_plugin: _BuiltPlugin): - activation = await plugin.activate_dynamic_plugins({}, [native_dynamic_plugin.spec()]) + activation = await plugin.initialize_with_dynamic_plugins({}, [native_dynamic_plugin.spec()]) assert "fixture_native" in plugin.list_kinds() del activation @@ -492,7 +492,7 @@ async def test_worker_activation_finalizer_never_waits_on_python_thread( ) ) - activation = await plugin.activate_dynamic_plugins( + activation = await plugin.initialize_with_dynamic_plugins( {}, [_BuiltPlugin("fixture_worker", "worker", manifest).spec()], ) @@ -533,7 +533,7 @@ async def test_worker_activation_finalizer_never_waits_on_python_thread( async def test_worker_activation_executes_and_releases_callbacks(worker_dynamic_plugin: _BuiltPlugin): - activation = await plugin.activate_dynamic_plugins({}, [worker_dynamic_plugin.spec()]) + activation = await plugin.initialize_with_dynamic_plugins({}, [worker_dynamic_plugin.spec()]) try: result = await tools.execute("python-worker-fixture", {"input": True}, lambda args: {"args": args}) assert result["worker_plugin_tool_execution"] is True