From 39ffe66270697d31abf503eddccfb2df3d1cf14e Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 6 Jul 2026 14:29:30 -0600 Subject: [PATCH 01/13] feat(node): activate dynamic plugins Signed-off-by: Bryan Bednarski --- crates/node/plugin.d.ts | 37 +++ crates/node/plugin.js | 14 + crates/node/src/api/mod.rs | 136 +++++++++ crates/node/tests/dynamic_plugin_tests.mjs | 323 +++++++++++++++++++++ 4 files changed, 510 insertions(+) create mode 100644 crates/node/tests/dynamic_plugin_tests.mjs diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index 76a84ee28..5802a6d2d 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -45,6 +45,28 @@ export interface PluginConfig { policy?: ConfigPolicy; } +/** Execution lane for a dynamically loaded Relay plugin. */ +export type DynamicPluginKind = 'rust_dynamic' | 'worker'; + +/** Explicitly resolved dynamic plugin load and component configuration. */ +export interface DynamicPluginActivationSpec { + pluginId: string; + kind: DynamicPluginKind; + manifestRef: string; + environmentRef?: string | null; + config?: Record; +} + +/** Owns one process-wide dynamic plugin host activation. */ +export interface DynamicPluginActivation { + /** Validation report produced by the successful activation. */ + readonly report: ConfigReport; + /** Whether this object still owns the dynamic plugin host. */ + readonly active: boolean; + /** Clear callbacks before unloading libraries and workers. Idempotent. */ + close(): Promise; +} + /** A mark Relay materializes under a managed lifecycle. */ export interface PendingMarkSpec { name: string; @@ -297,6 +319,21 @@ export declare function validate(config: PluginConfig): ConfigReport; * the promise rejects with the underlying validation or setup error. */ export declare function initialize(config: PluginConfig): Promise; +/** + * Load and activate explicitly resolved dynamic plugins. + * + * The returned object owns loaded libraries and worker processes. Keep it + * alive while plugin callbacks may run and call `close()` for deterministic + * teardown. Garbage collection is a defensive fallback only. + * + * @param config - Base plugin configuration activated alongside dynamic components. + * @param specs - Explicit manifest and component configuration for each plugin. + * @returns The owned activation and its validation report. + */ +export declare function activateDynamicPlugins( + config: PluginConfig, + specs: DynamicPluginActivationSpec[], +): Promise; /** * Clear the active plugin configuration. * diff --git a/crates/node/plugin.js b/crates/node/plugin.js index a84c3212b..7d1cfbeae 100644 --- a/crates/node/plugin.js +++ b/crates/node/plugin.js @@ -78,6 +78,19 @@ function initialize(config) { return lib.initializePlugins(config); } +/** + * Load and activate explicitly resolved dynamic plugins. + * + * @param {object} config - Base plugin configuration document. + * @param {Array} specs - Native-library or worker plugin load specifications. + * @returns {Promise} An owned activation with `report`, `active`, and `close()`. + * @remarks Keep the returned activation alive while its callbacks may run and + * call `close()` for deterministic teardown. + */ +function activateDynamicPlugins(config, specs) { + return lib.activateDynamicPlugins(config, specs); +} + /** * Clear the active plugin configuration. * @@ -161,6 +174,7 @@ module.exports = { ComponentSpec, validate, initialize, + activateDynamicPlugins, clear, report, listKinds, diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 9f06b649f..334c8869e 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -54,6 +54,10 @@ use nemo_relay::plugin::{ list_plugin_kinds as list_plugin_kinds_impl, register_plugin as register_plugin_impl, validate_plugin_config as validate_plugin_config_impl, }; +use nemo_relay::plugin::dynamic::{ + DynamicPluginActivationSpec as CoreDynamicPluginActivationSpec, DynamicPluginKind, + PluginHostActivation as CorePluginHostActivation, +}; use nemo_relay::shared_runtime::initialize_shared_runtime_binding; use nemo_relay_adaptive::acg::{ AgentIdentity, CacheRequestFacts, CacheTelemetryEvent, CacheTelemetryProvider, @@ -4007,6 +4011,138 @@ pub async fn initialize_plugins(config: Json) -> napi::Result { serde_json::to_value(&report).map_err(|e| napi::Error::from_reason(e.to_string())) } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct NodeDynamicPluginActivationSpec { + #[serde(alias = "plugin_id")] + plugin_id: String, + kind: DynamicPluginKind, + #[serde(alias = "manifest_ref")] + manifest_ref: String, + #[serde(default, alias = "environment_ref")] + environment_ref: Option, + #[serde(default)] + config: serde_json::Map, +} + +impl From for CoreDynamicPluginActivationSpec { + fn from(spec: NodeDynamicPluginActivationSpec) -> Self { + Self { + plugin_id: spec.plugin_id, + kind: spec.kind, + manifest_ref: spec.manifest_ref, + environment_ref: spec.environment_ref, + config: spec.config, + } + } +} + +/// Owned dynamic plugin activation. +/// +/// Keep this object alive while code may invoke callbacks registered by the +/// dynamic plugins. Call `close()` for deterministic cleanup; garbage +/// collection performs the same cleanup as a defensive fallback. +#[napi] +pub struct DynamicPluginActivation { + inner: Arc>>, + report: Json, +} + +#[napi] +impl DynamicPluginActivation { + /// Return the validation report produced by activation. + #[napi(getter)] + pub fn report(&self) -> Json { + self.report.clone() + } + + /// Return whether this object still owns the dynamic plugin host. + #[napi(getter)] + pub fn active(&self) -> napi::Result { + self.inner + .lock() + .map(|activation| activation.is_some()) + .map_err(|error| { + napi::Error::from_reason(format!( + "dynamic plugin activation lock poisoned: {error}" + )) + }) + } + + /// Clear plugin callbacks before unloading libraries and workers. + /// + /// This method is idempotent, including when concurrent callers race to + /// close the same activation. + #[napi(ts_return_type = "Promise")] + pub fn close(&self, env: Env) -> napi::Result { + let inner = Arc::clone(&self.inner); + env.execute_tokio_future( + async move { + let activation = inner + .lock() + .map_err(|error| { + napi::Error::from_reason(format!( + "dynamic plugin activation lock poisoned: {error}" + )) + })? + .take(); + if let Some(activation) = activation { + tokio::task::spawn_blocking(move || activation.clear()) + .await + .map_err(|error| { + napi::Error::from_reason(format!( + "dynamic plugin teardown task failed: {error}" + )) + })? + .map_err(|error| napi::Error::from_reason(error.to_string()))?; + } + Ok(()) + }, + |env, _| env.get_undefined(), + ) + } +} + +impl Drop for DynamicPluginActivation { + fn drop(&mut self) { + let activation = match self.inner.lock() { + Ok(mut activation) => activation.take(), + Err(poisoned) => poisoned.into_inner().take(), + }; + if let Some(activation) = activation + && let Err(error) = activation.clear() + { + eprintln!("nemo_relay: dynamic plugin finalizer teardown failed: {error}"); + } + } +} + +/// Load and activate explicitly resolved dynamic plugins. +/// +/// The returned object owns all loaded libraries and worker processes. Its +/// validation report is available through the `report` property. +#[napi] +pub async fn activate_dynamic_plugins( + config: Json, + specs: Json, +) -> napi::Result { + let config: PluginConfig = serde_json::from_value(config) + .map_err(|error| napi::Error::from_reason(format!("invalid plugin config: {error}")))?; + let specs: Vec = serde_json::from_value(specs).map_err(|error| { + napi::Error::from_reason(format!("invalid dynamic plugin specs: {error}")) + })?; + let specs = specs.into_iter().map(Into::into).collect::>(); + let (activation, report) = CorePluginHostActivation::activate(config, specs) + .await + .map_err(|error| napi::Error::from_reason(error.to_string()))?; + let report = serde_json::to_value(report) + .map_err(|error| napi::Error::from_reason(error.to_string()))?; + Ok(DynamicPluginActivation { + inner: Arc::new(StdMutex::new(Some(activation))), + report, + }) +} + /// Clear the active global plugin configuration. #[napi] pub fn clear_plugin_configuration() -> napi::Result<()> { diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs new file mode 100644 index 000000000..ecbbce066 --- /dev/null +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -0,0 +1,323 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { after, before, describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { execFileSync } from 'node:child_process'; +import { mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { createRequire } from 'node:module'; + +const require = createRequire(import.meta.url); +const lib = require('../index.js'); +const plugin = require('../plugin.js'); + +const nodeDir = fileURLToPath(new URL('..', import.meta.url)); +const repoRoot = path.resolve(nodeDir, '../..'); +const fixtureTarget = path.join(repoRoot, 'target', 'node-dynamic-plugin-fixtures'); +const tempRoot = mkdtempSync(path.join(tmpdir(), 'nemo-relay-node-dynamic-')); +const relayVersion = JSON.parse(readFileSync(path.join(nodeDir, 'package.json'), 'utf8')).version; + +let nativeManifestRef; +let workerManifestRef; + +function tomlString(value) { + return JSON.stringify(value); +} + +function nativeLibraryName() { + if (process.platform === 'win32') { + return 'nemo_relay_plugin_fixture.dll'; + } + if (process.platform === 'darwin') { + return 'libnemo_relay_plugin_fixture.dylib'; + } + return 'libnemo_relay_plugin_fixture.so'; +} + +function workerBinaryName() { + return process.platform === 'win32' ? 'nemo-relay-worker-plugin-fixture.exe' : 'nemo-relay-worker-plugin-fixture'; +} + +function buildFixture(manifestPath) { + execFileSync( + process.env.CARGO || 'cargo', + ['build', '--quiet', '--manifest-path', manifestPath, '--target-dir', fixtureTarget], + { stdio: 'inherit' }, + ); +} + +function buildNativeFixture(sourceManifestPath) { + const sourceDirectory = path.dirname(sourceManifestPath); + const fixtureDirectory = path.join(tempRoot, 'native-source'); + const fixtureSourceDirectory = path.join(fixtureDirectory, 'src'); + mkdirSync(fixtureSourceDirectory, { recursive: true }); + const pluginCrate = path.join(repoRoot, 'crates', 'plugin'); + const manifest = readFileSync(sourceManifestPath, 'utf8').replace( + 'nemo-relay-plugin = { path = "../../../../plugin" }', + `nemo-relay-plugin = { path = ${tomlString(pluginCrate)} }`, + ); + const fixtureManifest = path.join(fixtureDirectory, 'Cargo.toml'); + writeFileSync(fixtureManifest, manifest); + writeFileSync(path.join(fixtureSourceDirectory, 'lib.rs'), readFileSync(path.join(sourceDirectory, 'src', 'lib.rs'))); + buildFixture(fixtureManifest); +} + +function writeNativeManifest(libraryPath) { + const directory = path.join(tempRoot, 'native'); + mkdirSync(directory, { recursive: true }); + const manifestRef = path.join(directory, 'relay-plugin.toml'); + writeFileSync( + manifestRef, + `manifest_version = 1 + +[plugin] +id = "fixture_native" +kind = "rust_dynamic" + +[compat] +relay = ${tomlString(`=${relayVersion}`)} +native_api = "1" + +[defaults] +enabled = false + +[capabilities] +items = ["plugin_native"] + +[load] +library = ${tomlString(libraryPath)} +symbol = "nemo_relay_fixture_native_plugin" +`, + ); + return manifestRef; +} + +function writeWorkerManifest(entrypoint) { + const directory = path.join(tempRoot, 'worker'); + mkdirSync(directory, { recursive: true }); + const manifestRef = path.join(directory, 'relay-plugin.toml'); + writeFileSync( + manifestRef, + `manifest_version = 1 + +[plugin] +id = "fixture_worker" +kind = "worker" + +[compat] +relay = ${tomlString(`=${relayVersion}`)} +worker_protocol = "grpc-v1" + +[defaults] +enabled = false + +[capabilities] +items = ["plugin_worker"] + +[load] +runtime = "rust" +entrypoint = ${tomlString(entrypoint)} +`, + ); + return manifestRef; +} + +function activationSpec(pluginId, kind, manifestRef, config = {}) { + return { + pluginId, + kind, + manifestRef, + config, + }; +} + +function nativeRequest(model = 'fixture-model') { + return { + headers: {}, + content: { + model, + messages: [], + }, + }; +} + +async function executeTool(name) { + return lib.toolCallExecute( + name, + { original: true }, + (args) => ({ ...args, downstream: true }), + null, + null, + null, + null, + ); +} + +async function executeLlm(name) { + return lib.llmCallExecute( + name, + nativeRequest(), + (request) => ({ + downstream: true, + requestContent: request.content, + }), + null, + null, + null, + null, + null, + ); +} + +before(() => { + const nativeFixture = path.join(repoRoot, 'crates', 'core', 'tests', 'fixtures', 'native_plugin', 'Cargo.toml'); + const workerFixture = path.join(repoRoot, 'crates', 'core', 'tests', 'fixtures', 'worker_plugin', 'Cargo.toml'); + buildNativeFixture(nativeFixture); + buildFixture(workerFixture); + nativeManifestRef = writeNativeManifest(path.join(fixtureTarget, 'debug', nativeLibraryName())); + workerManifestRef = writeWorkerManifest(path.join(fixtureTarget, 'debug', workerBinaryName())); +}); + +after(() => { + rmSync(tempRoot, { recursive: true, force: true }); +}); + +describe('dynamic plugin host', () => { + it('owns native managed callbacks until idempotent close', async () => { + const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), + ]); + try { + assert.deepEqual(activation.report.diagnostics, []); + assert.equal(activation.active, true); + assert.throws(() => plugin.clear(), /active dynamic plugin host/i); + + const toolResult = await executeTool('node_native_dynamic_tool'); + assert.equal(toolResult.downstream, true); + assert.equal(toolResult.native_plugin_tool_execution_request, true); + assert.equal(toolResult.native_plugin_tool_execution, true); + + const llmResult = await executeLlm('node_native_dynamic_llm'); + assert.equal(llmResult.downstream, true); + assert.equal(llmResult.requestContent.native_plugin_llm_execution_request, true); + assert.equal(llmResult.native_plugin_llm_execution, true); + + await Promise.all([activation.close(), activation.close()]); + assert.equal(activation.active, false); + await activation.close(); + + const toolAfterClose = await executeTool('node_native_closed_tool'); + assert.deepEqual(toolAfterClose, { original: true, downstream: true }); + const llmAfterClose = await executeLlm('node_native_closed_llm'); + assert.equal(llmAfterClose.downstream, true); + assert.equal(llmAfterClose.requestContent.native_plugin_llm_execution_request, undefined); + assert.equal(llmAfterClose.native_plugin_llm_execution, undefined); + } finally { + await activation.close(); + } + }); + + it('owns worker managed callbacks until close', async () => { + const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_worker', 'worker', workerManifestRef), + ]); + try { + assert.deepEqual(activation.report.diagnostics, []); + const toolResult = await executeTool('node_worker_dynamic_tool'); + assert.equal(toolResult.worker_plugin_tool_execution_request, true); + assert.equal(toolResult.worker_plugin_tool_execution, true); + + const llmResult = await executeLlm('node_worker_dynamic_llm'); + assert.equal(llmResult.requestContent.worker_plugin_llm_execution_request, true); + assert.equal(llmResult.worker_plugin_llm_execution, true); + } finally { + await activation.close(); + } + + const toolAfterClose = await executeTool('node_worker_closed_tool'); + assert.deepEqual(toolAfterClose, { original: true, downstream: true }); + const llmAfterClose = await executeLlm('node_worker_closed_llm'); + assert.equal(llmAfterClose.requestContent.worker_plugin_llm_execution_request, undefined); + assert.equal(llmAfterClose.worker_plugin_llm_execution, undefined); + }); + + it('preserves manifest and validation diagnostics in rejected promises', async () => { + const missingManifest = path.join(tempRoot, 'missing', 'relay-plugin.toml'); + await assert.rejects( + () => + plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), + activationSpec('missing_native', 'rust_dynamic', missingManifest), + ]), + (error) => { + assert.match(error.message, /native plugin load failed/i); + assert.match(error.message, /relay-plugin\.toml/); + assert.match(error.message, /does not exist/i); + return true; + }, + ); + + await assert.rejects( + () => + plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef, { reject: true }), + ]), + /fixture rejection requested/i, + ); + + const recovered = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), + ]); + await recovered.close(); + }); + + it('defensively clears a native activation during garbage collection', () => { + const pluginModule = path.join(nodeDir, 'plugin.js'); + const script = ` + import { createRequire } from 'node:module'; + const require = createRequire(${JSON.stringify(path.join(nodeDir, 'package.json'))}); + const plugin = require(${JSON.stringify(pluginModule)}); + const config = { version: 1, components: [] }; + const specs = [${JSON.stringify(activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef))}]; + let activation = await plugin.activateDynamicPlugins(config, specs); + const weak = new WeakRef(activation); + activation = null; + let collected = false; + for (let index = 0; index < 100; index += 1) { + global.gc(); + await new Promise((resolve) => setImmediate(resolve)); + if (weak.deref() === undefined) { + collected = true; + break; + } + await new Promise((resolve) => setImmediate(resolve)); + } + if (!collected) { + throw new Error('dynamic activation was not garbage collected'); + } + let replacement; + let lastError; + for (let index = 0; index < 100; index += 1) { + global.gc(); + await new Promise((resolve) => setImmediate(resolve)); + try { + replacement = await plugin.activateDynamicPlugins(config, specs); + break; + } catch (error) { + lastError = error; + } + } + if (!replacement) { + throw lastError ?? new Error('dynamic activation finalizer did not release ownership'); + } + await replacement.close(); + `; + execFileSync(process.execPath, ['--expose-gc', '--input-type=module', '--eval', script], { + stdio: 'inherit', + timeout: 30_000, + }); + }); +}); From e83b050b1b362c93e186bbf7cdb4627200b23952 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 6 Jul 2026 18:13:39 -0600 Subject: [PATCH 02/13] Fix Node dynamic plugin teardown lifecycle Signed-off-by: Bryan Bednarski --- crates/node/src/api/mod.rs | 185 ++++++++++++++++----- crates/node/tests/dynamic_plugin_tests.mjs | 117 ++++++++++++- 2 files changed, 254 insertions(+), 48 deletions(-) diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 334c8869e..3258f71e8 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -46,6 +46,10 @@ use nemo_relay::api::tool::ToolAttributes; use nemo_relay::codec::request::AnnotatedLlmRequest; use nemo_relay::codec::response::Usage; use nemo_relay::error::{FlowError, Result as FlowResult}; +use nemo_relay::plugin::dynamic::{ + DynamicPluginActivationSpec as CoreDynamicPluginActivationSpec, DynamicPluginKind, + PluginHostActivation as CorePluginHostActivation, +}; use nemo_relay::plugin::{ ConfigDiagnostic, DiagnosticLevel, Plugin, PluginConfig, PluginError, PluginRegistration, PluginRegistrationContext, active_plugin_report as active_plugin_report_impl, @@ -54,10 +58,6 @@ use nemo_relay::plugin::{ list_plugin_kinds as list_plugin_kinds_impl, register_plugin as register_plugin_impl, validate_plugin_config as validate_plugin_config_impl, }; -use nemo_relay::plugin::dynamic::{ - DynamicPluginActivationSpec as CoreDynamicPluginActivationSpec, DynamicPluginKind, - PluginHostActivation as CorePluginHostActivation, -}; use nemo_relay::shared_runtime::initialize_shared_runtime_binding; use nemo_relay_adaptive::acg::{ AgentIdentity, CacheRequestFacts, CacheTelemetryEvent, CacheTelemetryProvider, @@ -4044,10 +4044,131 @@ impl From for CoreDynamicPluginActivationSpec { /// collection performs the same cleanup as a defensive fallback. #[napi] pub struct DynamicPluginActivation { - inner: Arc>>, + close_state: Arc, report: Json, } +type DynamicPluginTeardownResult = std::result::Result<(), String>; + +enum DynamicPluginCloseStatus { + Active(Option), + Closing, + Closed, +} + +struct DynamicPluginCloseState { + status: StdMutex, + completion: tokio::sync::watch::Sender>, +} + +impl DynamicPluginCloseState { + fn new(activation: CorePluginHostActivation) -> Self { + let (completion, _) = tokio::sync::watch::channel(None); + Self { + status: StdMutex::new(DynamicPluginCloseStatus::Active(Some(activation))), + completion, + } + } + + fn active(&self) -> bool { + let status = self + .status + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + match &*status { + DynamicPluginCloseStatus::Active(activation) => activation.is_some(), + DynamicPluginCloseStatus::Closing | DynamicPluginCloseStatus::Closed => false, + } + } + + fn begin_close(self: &Arc, log_finalizer_error: bool) { + let activation = { + let mut status = self + .status + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + match &mut *status { + DynamicPluginCloseStatus::Active(activation) => { + let activation = activation.take(); + *status = DynamicPluginCloseStatus::Closing; + activation + } + DynamicPluginCloseStatus::Closing | DynamicPluginCloseStatus::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 JS thread. + let activation = Arc::new(StdMutex::new(Some(activation))); + let worker_activation = Arc::clone(&activation); + let close_state = Arc::clone(self); + let spawn = std::thread::Builder::new() + .name("nemo-relay-node-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(|_| "dynamic plugin teardown task panicked".to_string()) + .and_then(|result| result.map_err(|error| error.to_string())) + } + None => Err("dynamic plugin teardown task lost its activation".to_string()), + }; + if log_finalizer_error && let Err(error) = &result { + eprintln!("nemo_relay: dynamic plugin finalizer teardown failed: {error}"); + } + close_state.finish(result); + }); + + if let Err(error) = spawn { + // Cleanup must never fall back to the JS 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); + } + let error = format!("failed to start dynamic plugin teardown task: {error}"); + if log_finalizer_error { + eprintln!("nemo_relay: dynamic plugin finalizer teardown failed: {error}"); + } + self.finish(Err(error)); + } + } + + fn finish(&self, result: DynamicPluginTeardownResult) { + *self + .status + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = DynamicPluginCloseStatus::Closed; + self.completion.send_replace(Some(result)); + } + + async fn wait_for_close(&self) -> DynamicPluginTeardownResult { + let mut completion = self.completion.subscribe(); + loop { + if let Some(result) = completion.borrow().clone() { + return result; + } + if completion.changed().await.is_err() { + return Err( + "dynamic plugin teardown result channel closed unexpectedly".to_string() + ); + } + } + } +} + #[napi] impl DynamicPluginActivation { /// Return the validation report produced by activation. @@ -4059,14 +4180,7 @@ impl DynamicPluginActivation { /// Return whether this object still owns the dynamic plugin host. #[napi(getter)] pub fn active(&self) -> napi::Result { - self.inner - .lock() - .map(|activation| activation.is_some()) - .map_err(|error| { - napi::Error::from_reason(format!( - "dynamic plugin activation lock poisoned: {error}" - )) - }) + Ok(self.close_state.active()) } /// Clear plugin callbacks before unloading libraries and workers. @@ -4075,28 +4189,14 @@ impl DynamicPluginActivation { /// close the same activation. #[napi(ts_return_type = "Promise")] pub fn close(&self, env: Env) -> napi::Result { - let inner = Arc::clone(&self.inner); + let close_state = Arc::clone(&self.close_state); + close_state.begin_close(false); env.execute_tokio_future( async move { - let activation = inner - .lock() - .map_err(|error| { - napi::Error::from_reason(format!( - "dynamic plugin activation lock poisoned: {error}" - )) - })? - .take(); - if let Some(activation) = activation { - tokio::task::spawn_blocking(move || activation.clear()) - .await - .map_err(|error| { - napi::Error::from_reason(format!( - "dynamic plugin teardown task failed: {error}" - )) - })? - .map_err(|error| napi::Error::from_reason(error.to_string()))?; - } - Ok(()) + close_state + .wait_for_close() + .await + .map_err(napi::Error::from_reason) }, |env, _| env.get_undefined(), ) @@ -4105,15 +4205,7 @@ impl DynamicPluginActivation { impl Drop for DynamicPluginActivation { fn drop(&mut self) { - let activation = match self.inner.lock() { - Ok(mut activation) => activation.take(), - Err(poisoned) => poisoned.into_inner().take(), - }; - if let Some(activation) = activation - && let Err(error) = activation.clear() - { - eprintln!("nemo_relay: dynamic plugin finalizer teardown failed: {error}"); - } + self.close_state.begin_close(true); } } @@ -4128,9 +4220,10 @@ pub async fn activate_dynamic_plugins( ) -> napi::Result { let config: PluginConfig = serde_json::from_value(config) .map_err(|error| napi::Error::from_reason(format!("invalid plugin config: {error}")))?; - let specs: Vec = serde_json::from_value(specs).map_err(|error| { - napi::Error::from_reason(format!("invalid dynamic plugin specs: {error}")) - })?; + let specs: Vec = + serde_json::from_value(specs).map_err(|error| { + napi::Error::from_reason(format!("invalid dynamic plugin specs: {error}")) + })?; let specs = specs.into_iter().map(Into::into).collect::>(); let (activation, report) = CorePluginHostActivation::activate(config, specs) .await @@ -4138,7 +4231,7 @@ pub async fn activate_dynamic_plugins( let report = serde_json::to_value(report) .map_err(|error| napi::Error::from_reason(error.to_string()))?; Ok(DynamicPluginActivation { - inner: Arc::new(StdMutex::new(Some(activation))), + close_state: Arc::new(DynamicPluginCloseState::new(activation)), report, }) } diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index ecbbce066..b8797a608 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -95,8 +95,8 @@ symbol = "nemo_relay_fixture_native_plugin" return manifestRef; } -function writeWorkerManifest(entrypoint) { - const directory = path.join(tempRoot, 'worker'); +function writeWorkerManifest(entrypoint, name = 'worker') { + const directory = path.join(tempRoot, name); mkdirSync(directory, { recursive: true }); const manifestRef = path.join(directory, 'relay-plugin.toml'); writeFileSync( @@ -125,6 +125,14 @@ entrypoint = ${tomlString(entrypoint)} return manifestRef; } +function writeWorkerWrapper(entrypoint, pidFile, name) { + const wrapper = path.join(tempRoot, `${name}.sh`); + writeFileSync(wrapper, `#!/bin/sh\nprintf '%s' "$$" > ${tomlString(pidFile)}\nexec ${tomlString(entrypoint)}\n`, { + mode: 0o755, + }); + return wrapper; +} + function activationSpec(pluginId, kind, manifestRef, config = {}) { return { pluginId, @@ -244,6 +252,46 @@ describe('dynamic plugin host', () => { assert.equal(llmAfterClose.worker_plugin_llm_execution, undefined); }); + it( + 'keeps every concurrent close pending until the shared teardown completes', + { skip: process.platform === 'win32' }, + async () => { + const workerBinary = path.join(fixtureTarget, 'debug', workerBinaryName()); + const pidFile = path.join(tempRoot, 'concurrent-close-worker.pid'); + const wrapper = writeWorkerWrapper(workerBinary, pidFile, 'concurrent-close-worker'); + const manifestRef = writeWorkerManifest(wrapper, 'concurrent-close-worker'); + const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_worker', 'worker', manifestRef), + ]); + const workerPid = Number(readFileSync(pidFile, 'utf8')); + process.kill(workerPid, 'SIGSTOP'); + + const firstClose = activation.close(); + const secondClose = activation.close(); + let earlyResult; + try { + earlyResult = await Promise.race([ + firstClose.then(() => 'first'), + secondClose.then(() => 'second'), + new Promise((resolve) => setTimeout(() => resolve('pending'), 200)), + ]); + } finally { + try { + process.kill(workerPid, 'SIGCONT'); + } catch (error) { + if (error.code !== 'ESRCH') { + throw error; + } + } + await Promise.all([firstClose, secondClose]); + } + + assert.equal(earlyResult, 'pending'); + assert.equal(activation.active, false); + await activation.close(); + }, + ); + it('preserves manifest and validation diagnostics in rejected promises', async () => { const missingManifest = path.join(tempRoot, 'missing', 'relay-plugin.toml'); await assert.rejects( @@ -320,4 +368,69 @@ describe('dynamic plugin host', () => { timeout: 30_000, }); }); + + it( + 'never waits for worker teardown on the JavaScript thread during garbage collection', + { skip: process.platform === 'win32' }, + () => { + const workerBinary = path.join(fixtureTarget, 'debug', workerBinaryName()); + const pidFile = path.join(tempRoot, 'finalizer-worker.pid'); + const wrapper = writeWorkerWrapper(workerBinary, pidFile, 'finalizer-worker'); + const manifestRef = writeWorkerManifest(wrapper, 'finalizer-worker'); + const pluginModule = path.join(nodeDir, 'plugin.js'); + const script = ` + import { spawn } from 'node:child_process'; + import { readFileSync } from 'node:fs'; + import { performance } from 'node:perf_hooks'; + import { createRequire } from 'node:module'; + const require = createRequire(${JSON.stringify(path.join(nodeDir, 'package.json'))}); + const plugin = require(${JSON.stringify(pluginModule)}); + const config = { version: 1, components: [] }; + const specs = [${JSON.stringify(activationSpec('fixture_worker', 'worker', manifestRef))}]; + let activation = await plugin.activateDynamicPlugins(config, specs); + const weak = new WeakRef(activation); + activation = null; + await new Promise((resolve) => setImmediate(resolve)); + + const workerPid = Number(readFileSync(${JSON.stringify(pidFile)}, 'utf8')); + process.kill(workerPid, 'SIGSTOP'); + const resumer = spawn( + '/bin/sh', + ['-c', 'sleep 0.8; kill -CONT "$1"', 'resume-worker', String(workerPid)], + { detached: true, stdio: 'ignore' }, + ); + resumer.unref(); + + const startedAt = performance.now(); + global.gc(); + const elapsed = performance.now() - startedAt; + if (weak.deref() !== undefined) { + throw new Error('dynamic activation was not garbage collected'); + } + if (elapsed >= 400) { + throw new Error(\`dynamic activation finalizer blocked the JavaScript thread for \${elapsed}ms\`); + } + + let replacement; + let lastError; + for (let index = 0; index < 500; index += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + try { + replacement = await plugin.activateDynamicPlugins(config, specs); + break; + } catch (error) { + lastError = error; + } + } + if (!replacement) { + throw lastError ?? new Error('dynamic activation finalizer did not release ownership'); + } + await replacement.close(); + `; + execFileSync(process.execPath, ['--expose-gc', '--input-type=module', '--eval', script], { + stdio: 'inherit', + timeout: 30_000, + }); + }, + ); }); From 9602c63f7d446610b1b7b875b424732ddc8cce59 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 7 Jul 2026 13:50:01 -0600 Subject: [PATCH 03/13] fix(node): preserve static-only plugin activation Signed-off-by: Bryan Bednarski --- crates/node/plugin.d.ts | 5 +-- crates/node/plugin.js | 7 ++-- crates/node/src/api/mod.rs | 7 ++-- crates/node/tests/dynamic_plugin_tests.mjs | 38 ++++++++++++++++++++++ 4 files changed, 50 insertions(+), 7 deletions(-) diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index 5802a6d2d..2e9f5127e 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -326,9 +326,10 @@ export declare function initialize(config: PluginConfig): Promise; * alive while plugin callbacks may run and call `close()` for deterministic * teardown. Garbage collection is a defensive fallback only. * - * @param config - Base plugin configuration activated alongside dynamic components. - * @param specs - Explicit manifest and component configuration for each plugin. + * @param config - Base configuration activated first, including any static components. + * @param specs - Non-empty explicit manifest and component configuration for each plugin. * @returns The owned activation and its validation report. + * @remarks Use `initialize()` for a static-only configuration. */ export declare function activateDynamicPlugins( config: PluginConfig, diff --git a/crates/node/plugin.js b/crates/node/plugin.js index 7d1cfbeae..af24af161 100644 --- a/crates/node/plugin.js +++ b/crates/node/plugin.js @@ -81,11 +81,12 @@ function initialize(config) { /** * Load and activate explicitly resolved dynamic plugins. * - * @param {object} config - Base plugin configuration document. - * @param {Array} specs - Native-library or worker plugin load specifications. + * @param {object} config - Base configuration, including any static components. + * @param {Array} specs - Non-empty native-library or worker plugin specifications. * @returns {Promise} An owned activation with `report`, `active`, and `close()`. * @remarks Keep the returned activation alive while its callbacks may run and - * call `close()` for deterministic teardown. + * call `close()` for deterministic teardown. Use `initialize()` for a + * static-only configuration. */ function activateDynamicPlugins(config, specs) { return lib.activateDynamicPlugins(config, specs); diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 3258f71e8..f45f2ab6c 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -4211,8 +4211,11 @@ impl Drop for DynamicPluginActivation { /// Load and activate explicitly resolved dynamic plugins. /// -/// The returned object owns all loaded libraries and worker processes. Its -/// validation report is available through the `report` property. +/// `config` may contain statically registered components; dynamic components +/// are activated after that base configuration. At least one dynamic plugin is +/// required. Static-only callers should use `initializePlugins`. The returned +/// object owns all loaded libraries and worker processes. Its validation report +/// is available through the `report` property. #[napi] pub async fn activate_dynamic_plugins( config: Json, diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index b8797a608..dfa046aa3 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -194,6 +194,44 @@ after(() => { }); describe('dynamic plugin host', () => { + it('rejects empty specs without taking over static initialization', async () => { + await assert.rejects( + () => plugin.activateDynamicPlugins({ version: 1, components: [] }, []), + /at least one dynamic plugin/i, + ); + + assert.deepEqual(await plugin.initialize({ version: 1, components: [] }), { diagnostics: [] }); + plugin.clear(); + }); + + it('activates static base components with dynamic plugins', async () => { + const staticKind = 'node.fixture.static-base'; + plugin.register(staticKind, { + register(_config, context) { + context.registerToolRequestIntercept('mark-static-base', 0, false, (_name, args) => ({ + ...args, + staticBase: true, + })); + }, + }); + let activation; + try { + activation = await plugin.activateDynamicPlugins( + { + version: 1, + components: [{ kind: staticKind, enabled: true, config: {} }], + }, + [activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef)], + ); + const result = await executeTool('node_static_and_dynamic_tool'); + assert.equal(result.staticBase, true); + assert.equal(result.native_plugin_tool_execution, true); + } finally { + await activation?.close(); + plugin.deregister(staticKind); + } + }); + it('owns native managed callbacks until idempotent close', async () => { const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), From 16c509625fa892ae14560d463a94af3414d68957 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 13 Jul 2026 21:06:15 -0600 Subject: [PATCH 04/13] docs(node): clarify plugin activation state Signed-off-by: Bryan Bednarski --- crates/node/plugin.d.ts | 5 ++++- crates/node/src/api/mod.rs | 5 ++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index 2e9f5127e..e5ca98775 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -61,7 +61,10 @@ export interface DynamicPluginActivationSpec { export interface DynamicPluginActivation { /** Validation report produced by the successful activation. */ readonly report: ConfigReport; - /** Whether this object still owns the dynamic plugin host. */ + /** + * Whether this activation handle has not begun teardown. `false` does not + * guarantee another process-wide activation can start after failed teardown. + */ readonly active: boolean; /** Clear callbacks before unloading libraries and workers. Idempotent. */ close(): Promise; diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index f45f2ab6c..f00e54919 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -4177,7 +4177,10 @@ impl DynamicPluginActivation { self.report.clone() } - /// Return whether this object still owns the 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. #[napi(getter)] pub fn active(&self) -> napi::Result { Ok(self.close_state.active()) From b43380b269264a38b07f769a4e0456fb8b596d83 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 13 Jul 2026 23:00:12 -0600 Subject: [PATCH 05/13] feat(node): support structured plugin activation disposal Signed-off-by: Bryan Bednarski --- crates/node/plugin.d.ts | 4 +++- crates/node/plugin.js | 6 +++--- crates/node/src/api/mod.rs | 22 ++++++++++++++++++++-- crates/node/tests/dynamic_plugin_tests.mjs | 20 ++++++++++++++++++++ 4 files changed, 46 insertions(+), 6 deletions(-) diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index e5ca98775..83a62bb92 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -58,7 +58,7 @@ export interface DynamicPluginActivationSpec { } /** Owns one process-wide dynamic plugin host activation. */ -export interface DynamicPluginActivation { +export interface DynamicPluginActivation extends AsyncDisposable { /** Validation report produced by the successful activation. */ readonly report: ConfigReport; /** @@ -68,6 +68,8 @@ export interface DynamicPluginActivation { readonly active: boolean; /** Clear callbacks before unloading libraries and workers. Idempotent. */ close(): Promise; + /** Delegate structured `await using` cleanup to `close()`. */ + [Symbol.asyncDispose](): Promise; } /** A mark Relay materializes under a managed lifecycle. */ diff --git a/crates/node/plugin.js b/crates/node/plugin.js index af24af161..32162120a 100644 --- a/crates/node/plugin.js +++ b/crates/node/plugin.js @@ -83,10 +83,10 @@ function initialize(config) { * * @param {object} config - Base configuration, including any static components. * @param {Array} specs - Non-empty native-library or worker plugin specifications. - * @returns {Promise} An owned activation with `report`, `active`, and `close()`. + * @returns {Promise} An owned activation with `report`, `active`, `close()`, and async disposal. * @remarks Keep the returned activation alive while its callbacks may run and - * call `close()` for deterministic teardown. Use `initialize()` for a - * static-only configuration. + * call `close()` or use `await using` for deterministic teardown. Use + * `initialize()` for a static-only configuration. */ function activateDynamicPlugins(config, specs) { return lib.activateDynamicPlugins(config, specs); diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index f00e54919..8795e0672 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -21,8 +21,8 @@ use std::sync::atomic::{AtomicU64, Ordering}; use chrono::{DateTime, Utc}; use napi::bindgen_prelude::*; use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode}; -use napi::{JsFunction, JsObject, JsUnknown, NapiRaw, NapiValue}; -use napi_derive::napi; +use napi::{JsFunction, JsObject, JsSymbol, JsUnknown, NapiRaw, NapiValue}; +use napi_derive::{module_exports, napi}; use serde::Deserialize; use serde_json::Value as Json; use tokio_stream::StreamExt; @@ -89,6 +89,18 @@ fn init() { .expect("node pii redaction plugin component registration should succeed"); } +#[module_exports] +fn install_well_known_symbol_methods(exports: JsObject, env: Env) -> napi::Result<()> { + let activation: JsFunction = exports.get_named_property("DynamicPluginActivation")?; + let activation = activation.coerce_to_object()?; + let mut prototype: JsObject = activation.get_named_property("prototype")?; + let symbol: JsFunction = env.get_global()?.get_named_property("Symbol")?; + let symbol = symbol.coerce_to_object()?; + let async_dispose: JsSymbol = symbol.get_named_property("asyncDispose")?; + let close: JsFunction = prototype.get_named_property("close")?; + prototype.set_property(async_dispose, close) +} + fn parse_string_map( value: Option, field_name: &str, @@ -4204,6 +4216,12 @@ impl DynamicPluginActivation { |env, _| env.get_undefined(), ) } + + /// Delegate structured `await using` cleanup to `close()`. + #[napi(js_name = "[Symbol.asyncDispose]", ts_return_type = "Promise")] + pub fn async_dispose(&self, env: Env) -> napi::Result { + self.close(env) + } } impl Drop for DynamicPluginActivation { diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index dfa046aa3..f42075b40 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -266,6 +266,26 @@ describe('dynamic plugin host', () => { } }); + it('supports structured async disposal when the managed scope throws', async () => { + let disposedActivation; + await assert.rejects(async () => { + await using activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), + ]); + disposedActivation = activation; + + assert.equal(activation[Symbol.asyncDispose], lib.DynamicPluginActivation.prototype.close); + const toolResult = await executeTool('node_native_async_dispose_tool'); + assert.equal(toolResult.native_plugin_tool_execution, true); + throw new Error('managed activation scope failed'); + }, /managed activation scope failed/); + + assert.equal(disposedActivation.active, false); + await disposedActivation[Symbol.asyncDispose](); + const toolAfterDispose = await executeTool('node_native_async_disposed_tool'); + assert.deepEqual(toolAfterDispose, { original: true, downstream: true }); + }); + it('owns worker managed callbacks until close', async () => { const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_worker', 'worker', workerManifestRef), From 66c6f2360e33d73c13b6a3a6edaaae3bd6d7ebd6 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Mon, 13 Jul 2026 23:29:37 -0600 Subject: [PATCH 06/13] fix(node): layer discovered plugin configuration Signed-off-by: Bryan Bednarski --- crates/node/plugin.d.ts | 5 +-- crates/node/plugin.js | 7 +++-- crates/node/src/api/mod.rs | 18 ++++++----- crates/node/tests/dynamic_plugin_tests.mjs | 36 +++++++++++++++++----- 4 files changed, 45 insertions(+), 21 deletions(-) diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index 83a62bb92..93f00d22c 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -331,10 +331,11 @@ export declare function initialize(config: PluginConfig): Promise; * alive while plugin callbacks may run and call `close()` for deterministic * teardown. Garbage collection is a defensive fallback only. * - * @param config - Base configuration activated first, including any static components. + * @param config - Base configuration layered over discovered `plugins.toml` files. * @param specs - Non-empty explicit manifest and component configuration for each plugin. * @returns The owned activation and its validation report. - * @remarks Use `initialize()` for a static-only configuration. + * @remarks File-configured static components initialize before dynamic + * components. Use `initialize()` for a static-only configuration. */ export declare function activateDynamicPlugins( config: PluginConfig, diff --git a/crates/node/plugin.js b/crates/node/plugin.js index 32162120a..bc99fc155 100644 --- a/crates/node/plugin.js +++ b/crates/node/plugin.js @@ -81,11 +81,12 @@ function initialize(config) { /** * Load and activate explicitly resolved dynamic plugins. * - * @param {object} config - Base configuration, including any static components. + * @param {object} config - Base configuration layered over discovered `plugins.toml` files. * @param {Array} specs - Non-empty native-library or worker plugin specifications. * @returns {Promise} An owned activation with `report`, `active`, `close()`, and async disposal. - * @remarks Keep the returned activation alive while its callbacks may run and - * call `close()` or use `await using` for deterministic teardown. Use + * @remarks File-configured static components initialize before dynamic + * components. Keep the returned activation alive while its callbacks may run + * and call `close()` or use `await using` for deterministic teardown. Use * `initialize()` for a static-only configuration. */ function activateDynamicPlugins(config, specs) { diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 8795e0672..93437514c 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -4232,11 +4232,12 @@ impl Drop for DynamicPluginActivation { /// Load and activate explicitly resolved dynamic plugins. /// -/// `config` may contain statically registered components; dynamic components -/// are activated after that base configuration. At least one dynamic plugin is -/// required. Static-only callers should use `initializePlugins`. The returned -/// object owns all loaded libraries and worker processes. Its validation report -/// is available through the `report` property. +/// `config` is layered over discovered `plugins.toml` files and may contain +/// statically registered components; dynamic components are activated after +/// that effective base configuration. At least one dynamic plugin is required. +/// Static-only callers should use `initializePlugins`. The returned object owns +/// all loaded libraries and worker processes. Its validation report is available +/// through the `report` property. #[napi] pub async fn activate_dynamic_plugins( config: Json, @@ -4249,9 +4250,10 @@ pub async fn activate_dynamic_plugins( napi::Error::from_reason(format!("invalid dynamic plugin specs: {error}")) })?; let specs = specs.into_iter().map(Into::into).collect::>(); - let (activation, report) = CorePluginHostActivation::activate(config, specs) - .await - .map_err(|error| napi::Error::from_reason(error.to_string()))?; + let (activation, report) = + CorePluginHostActivation::activate_with_discovered_config(config, specs) + .await + .map_err(|error| napi::Error::from_reason(error.to_string()))?; let report = serde_json::to_value(report) .map_err(|error| napi::Error::from_reason(error.to_string()))?; Ok(DynamicPluginActivation { diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index f42075b40..5434a06af 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -204,8 +204,22 @@ describe('dynamic plugin host', () => { plugin.clear(); }); - it('activates static base components with dynamic plugins', async () => { + it('layers plugins.toml static base components with dynamic plugins', async () => { const staticKind = 'node.fixture.static-base'; + const projectRoot = path.join(tempRoot, 'file-static-base-project'); + const projectConfigDirectory = path.join(projectRoot, '.nemo-relay'); + const isolatedUserConfig = path.join(projectRoot, 'xdg'); + mkdirSync(projectConfigDirectory, { recursive: true }); + mkdirSync(isolatedUserConfig, { recursive: true }); + writeFileSync( + path.join(projectConfigDirectory, 'plugins.toml'), + `version = 1 + +[[components]] +kind = ${tomlString(staticKind)} +enabled = true +`, + ); plugin.register(staticKind, { register(_config, context) { context.registerToolRequestIntercept('mark-static-base', 0, false, (_name, args) => ({ @@ -214,21 +228,27 @@ describe('dynamic plugin host', () => { })); }, }); + const previousCwd = process.cwd(); + const previousXdgConfigHome = process.env.XDG_CONFIG_HOME; let activation; try { - activation = await plugin.activateDynamicPlugins( - { - version: 1, - components: [{ kind: staticKind, enabled: true, config: {} }], - }, - [activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef)], - ); + process.chdir(projectRoot); + process.env.XDG_CONFIG_HOME = isolatedUserConfig; + activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), + ]); const result = await executeTool('node_static_and_dynamic_tool'); assert.equal(result.staticBase, true); assert.equal(result.native_plugin_tool_execution, true); } finally { await activation?.close(); plugin.deregister(staticKind); + process.chdir(previousCwd); + if (previousXdgConfigHome === undefined) { + delete process.env.XDG_CONFIG_HOME; + } else { + process.env.XDG_CONFIG_HOME = previousXdgConfigHome; + } } }); From 26ca19680e379366dd9995dccf20e013a7706d4f Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 00:32:58 -0600 Subject: [PATCH 07/13] test(node): preserve concurrent close failures Signed-off-by: Bryan Bednarski --- crates/node/tests/dynamic_plugin_tests.mjs | 23 ++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index 5434a06af..e083ab12f 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -347,21 +347,40 @@ enabled = true const firstClose = activation.close(); const secondClose = activation.close(); let earlyResult; + let operationFailed = false; + let operationError; try { earlyResult = await Promise.race([ firstClose.then(() => 'first'), secondClose.then(() => 'second'), new Promise((resolve) => setTimeout(() => resolve('pending'), 200)), ]); + } catch (error) { + operationFailed = true; + operationError = error; } finally { + let cleanupFailed = false; + let cleanupError; try { process.kill(workerPid, 'SIGCONT'); } catch (error) { if (error.code !== 'ESRCH') { - throw error; + cleanupFailed = true; + cleanupError = error; } } - await Promise.all([firstClose, secondClose]); + const closeResults = await Promise.allSettled([firstClose, secondClose]); + const rejectedClose = closeResults.find((result) => result.status === 'rejected'); + if (!cleanupFailed && rejectedClose !== undefined) { + cleanupFailed = true; + cleanupError = rejectedClose.reason; + } + if (operationFailed) { + throw operationError; + } + if (cleanupFailed) { + throw cleanupError; + } } assert.equal(earlyResult, 'pending'); From f8a3f31116322bab9642ce54c2f798d622e1a994 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 00:36:15 -0600 Subject: [PATCH 08/13] fix(node): declare async disposal library Signed-off-by: Bryan Bednarski --- crates/node/plugin.d.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index 93f00d22c..de9d54af5 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -1,6 +1,8 @@ // SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +/// + import type { EventSanitizeFields, Json } from './index'; /** Policy behavior for unsupported configuration. */ From f80aa2e82a1e68a4dd0946ab7396b1f824006e1d Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 00:39:39 -0600 Subject: [PATCH 09/13] fix(node): hide async disposal declaration shim Signed-off-by: Bryan Bednarski --- crates/node/src/api/mod.rs | 9 +++++++-- crates/node/tests/dynamic_plugin_tests.mjs | 1 + 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 93437514c..9e58bea27 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -98,7 +98,9 @@ fn install_well_known_symbol_methods(exports: JsObject, env: Env) -> napi::Resul let symbol = symbol.coerce_to_object()?; let async_dispose: JsSymbol = symbol.get_named_property("asyncDispose")?; let close: JsFunction = prototype.get_named_property("close")?; - prototype.set_property(async_dispose, close) + prototype.set_property(async_dispose, close)?; + prototype.delete_named_property("[Symbol.asyncDispose]")?; + Ok(()) } fn parse_string_map( @@ -4217,7 +4219,10 @@ impl DynamicPluginActivation { ) } - /// Delegate structured `await using` cleanup to `close()`. + /// Supply the structured disposal signature to napi-rs declaration generation. + /// + /// Module initialization installs `close()` under the actual well-known + /// symbol and removes this string-named declaration shim from the prototype. #[napi(js_name = "[Symbol.asyncDispose]", ts_return_type = "Promise")] pub fn async_dispose(&self, env: Env) -> napi::Result { self.close(env) diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index e083ab12f..0e33e1571 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -295,6 +295,7 @@ enabled = true disposedActivation = activation; assert.equal(activation[Symbol.asyncDispose], lib.DynamicPluginActivation.prototype.close); + assert.equal('[Symbol.asyncDispose]' in lib.DynamicPluginActivation.prototype, false); const toolResult = await executeTool('node_native_async_dispose_tool'); assert.equal(toolResult.native_plugin_tool_execution, true); throw new Error('managed activation scope failed'); From 4f99477c92f15bf65c34a849aae9abe7984bc21d Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 00:41:17 -0600 Subject: [PATCH 10/13] test(node): tolerate delayed activation collection Signed-off-by: Bryan Bednarski --- crates/node/tests/dynamic_plugin_tests.mjs | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index 0e33e1571..6040073d1 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -499,14 +499,22 @@ enabled = true ); resumer.unref(); - const startedAt = performance.now(); - global.gc(); - const elapsed = performance.now() - startedAt; - if (weak.deref() !== undefined) { - throw new Error('dynamic activation was not garbage collected'); + let collected = false; + for (let index = 0; index < 20; index += 1) { + const startedAt = performance.now(); + global.gc(); + const elapsed = performance.now() - startedAt; + if (elapsed >= 400) { + throw new Error(\`dynamic activation finalizer blocked the JavaScript thread for \${elapsed}ms\`); + } + if (weak.deref() === undefined) { + collected = true; + break; + } + await new Promise((resolve) => setImmediate(resolve)); } - if (elapsed >= 400) { - throw new Error(\`dynamic activation finalizer blocked the JavaScript thread for \${elapsed}ms\`); + if (!collected) { + throw new Error('dynamic activation was not garbage collected'); } let replacement; From 8735e6d8d655f9f88fc50295444bc5a4572339d2 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 07:18:49 -0600 Subject: [PATCH 11/13] fix(node): close remaining activation review gaps Signed-off-by: Bryan Bednarski --- crates/node/plugin.js | 11 ++++++++++- crates/node/tests/dynamic_plugin_tests.mjs | 16 ++++++++-------- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/crates/node/plugin.js b/crates/node/plugin.js index bc99fc155..33bbb264e 100644 --- a/crates/node/plugin.js +++ b/crates/node/plugin.js @@ -78,11 +78,20 @@ function initialize(config) { return lib.initializePlugins(config); } +/** + * @typedef {object} DynamicPluginActivationSpec + * @property {string} pluginId - Manifest plugin identifier. + * @property {'rust_dynamic'|'worker'} kind - Dynamic plugin execution kind. + * @property {string} manifestRef - Path to the plugin manifest. + * @property {string|null} [environmentRef] - Optional worker environment path. + * @property {Object} [config] - Plugin component configuration. + */ + /** * Load and activate explicitly resolved dynamic plugins. * * @param {object} config - Base configuration layered over discovered `plugins.toml` files. - * @param {Array} specs - Non-empty native-library or worker plugin specifications. + * @param {DynamicPluginActivationSpec[]} specs - Non-empty native-library or worker plugin specifications. * @returns {Promise} An owned activation with `report`, `active`, `close()`, and async disposal. * @remarks File-configured static components initialize before dynamic * components. Keep the returned activation alive while its callbacks may run diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index 6040073d1..19b9018b2 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -350,6 +350,8 @@ enabled = true let earlyResult; let operationFailed = false; let operationError; + let cleanupFailed = false; + let cleanupError; try { earlyResult = await Promise.race([ firstClose.then(() => 'first'), @@ -360,8 +362,6 @@ enabled = true operationFailed = true; operationError = error; } finally { - let cleanupFailed = false; - let cleanupError; try { process.kill(workerPid, 'SIGCONT'); } catch (error) { @@ -376,12 +376,12 @@ enabled = true cleanupFailed = true; cleanupError = rejectedClose.reason; } - if (operationFailed) { - throw operationError; - } - if (cleanupFailed) { - throw cleanupError; - } + } + if (operationFailed) { + throw operationError; + } + if (cleanupFailed) { + throw cleanupError; } assert.equal(earlyResult, 'pending'); From 4d3517fa79c33f698273e5d5cc9f34c3c950903a Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 08:02:53 -0600 Subject: [PATCH 12/13] fix(node): exclude runtime export hook from lib tests Signed-off-by: Bryan Bednarski --- crates/node/src/api/mod.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 9e58bea27..01486cc76 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -21,8 +21,8 @@ use std::sync::atomic::{AtomicU64, Ordering}; use chrono::{DateTime, Utc}; use napi::bindgen_prelude::*; use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode}; -use napi::{JsFunction, JsObject, JsSymbol, JsUnknown, NapiRaw, NapiValue}; -use napi_derive::{module_exports, napi}; +use napi::{JsFunction, JsObject, JsUnknown, NapiRaw, NapiValue}; +use napi_derive::napi; use serde::Deserialize; use serde_json::Value as Json; use tokio_stream::StreamExt; @@ -89,14 +89,15 @@ fn init() { .expect("node pii redaction plugin component registration should succeed"); } -#[module_exports] +#[cfg(not(test))] +#[napi_derive::module_exports] fn install_well_known_symbol_methods(exports: JsObject, env: Env) -> napi::Result<()> { let activation: JsFunction = exports.get_named_property("DynamicPluginActivation")?; let activation = activation.coerce_to_object()?; let mut prototype: JsObject = activation.get_named_property("prototype")?; let symbol: JsFunction = env.get_global()?.get_named_property("Symbol")?; let symbol = symbol.coerce_to_object()?; - let async_dispose: JsSymbol = symbol.get_named_property("asyncDispose")?; + let async_dispose: napi::JsSymbol = symbol.get_named_property("asyncDispose")?; let close: JsFunction = prototype.get_named_property("close")?; prototype.set_property(async_dispose, close)?; prototype.delete_named_property("[Symbol.asyncDispose]")?; From 12e182a918b04a6edfd5f50d62083a823664cd55 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 09:12:58 -0600 Subject: [PATCH 13/13] refactor(node): clarify dynamic plugin initialization Signed-off-by: Bryan Bednarski --- crates/node/plugin.d.ts | 4 ++-- crates/node/plugin.js | 8 +++---- crates/node/src/api/mod.rs | 4 ++-- crates/node/tests/dynamic_plugin_tests.mjs | 26 +++++++++++----------- 4 files changed, 21 insertions(+), 21 deletions(-) diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index de9d54af5..608289762 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -327,7 +327,7 @@ export declare function validate(config: PluginConfig): ConfigReport; */ export declare function initialize(config: PluginConfig): Promise; /** - * Load and activate explicitly resolved dynamic plugins. + * Initialize with explicitly resolved dynamic plugins. * * The returned object owns loaded libraries and worker processes. Keep it * alive while plugin callbacks may run and call `close()` for deterministic @@ -339,7 +339,7 @@ export declare function initialize(config: PluginConfig): Promise; * @remarks File-configured static components initialize before dynamic * components. Use `initialize()` for a static-only configuration. */ -export declare function activateDynamicPlugins( +export declare function initializeWithDynamicPlugins( config: PluginConfig, specs: DynamicPluginActivationSpec[], ): Promise; diff --git a/crates/node/plugin.js b/crates/node/plugin.js index 33bbb264e..84ab6d228 100644 --- a/crates/node/plugin.js +++ b/crates/node/plugin.js @@ -88,7 +88,7 @@ function initialize(config) { */ /** - * Load and activate explicitly resolved dynamic plugins. + * Initialize with explicitly resolved dynamic plugins. * * @param {object} config - Base configuration layered over discovered `plugins.toml` files. * @param {DynamicPluginActivationSpec[]} specs - Non-empty native-library or worker plugin specifications. @@ -98,8 +98,8 @@ function initialize(config) { * and call `close()` or use `await using` for deterministic teardown. Use * `initialize()` for a static-only configuration. */ -function activateDynamicPlugins(config, specs) { - return lib.activateDynamicPlugins(config, specs); +function initializeWithDynamicPlugins(config, specs) { + return lib.initializeWithDynamicPlugins(config, specs); } /** @@ -185,7 +185,7 @@ module.exports = { ComponentSpec, validate, initialize, - activateDynamicPlugins, + initializeWithDynamicPlugins, clear, report, listKinds, diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 01486cc76..ad9204973 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -4236,7 +4236,7 @@ impl Drop for DynamicPluginActivation { } } -/// Load and activate explicitly resolved dynamic plugins. +/// Initialize with explicitly resolved dynamic plugins. /// /// `config` is layered over discovered `plugins.toml` files and may contain /// statically registered components; dynamic components are activated after @@ -4245,7 +4245,7 @@ impl Drop for DynamicPluginActivation { /// all loaded libraries and worker processes. Its validation report is available /// through the `report` property. #[napi] -pub async fn activate_dynamic_plugins( +pub async fn initialize_with_dynamic_plugins( config: Json, specs: Json, ) -> napi::Result { diff --git a/crates/node/tests/dynamic_plugin_tests.mjs b/crates/node/tests/dynamic_plugin_tests.mjs index 19b9018b2..bf96f4387 100644 --- a/crates/node/tests/dynamic_plugin_tests.mjs +++ b/crates/node/tests/dynamic_plugin_tests.mjs @@ -196,7 +196,7 @@ after(() => { describe('dynamic plugin host', () => { it('rejects empty specs without taking over static initialization', async () => { await assert.rejects( - () => plugin.activateDynamicPlugins({ version: 1, components: [] }, []), + () => plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, []), /at least one dynamic plugin/i, ); @@ -234,7 +234,7 @@ enabled = true try { process.chdir(projectRoot); process.env.XDG_CONFIG_HOME = isolatedUserConfig; - activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + activation = await plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), ]); const result = await executeTool('node_static_and_dynamic_tool'); @@ -253,7 +253,7 @@ enabled = true }); it('owns native managed callbacks until idempotent close', async () => { - const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + const activation = await plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), ]); try { @@ -289,7 +289,7 @@ enabled = true it('supports structured async disposal when the managed scope throws', async () => { let disposedActivation; await assert.rejects(async () => { - await using activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + await using activation = await plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), ]); disposedActivation = activation; @@ -308,7 +308,7 @@ enabled = true }); it('owns worker managed callbacks until close', async () => { - const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + const activation = await plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_worker', 'worker', workerManifestRef), ]); try { @@ -339,7 +339,7 @@ enabled = true const pidFile = path.join(tempRoot, 'concurrent-close-worker.pid'); const wrapper = writeWorkerWrapper(workerBinary, pidFile, 'concurrent-close-worker'); const manifestRef = writeWorkerManifest(wrapper, 'concurrent-close-worker'); - const activation = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + const activation = await plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_worker', 'worker', manifestRef), ]); const workerPid = Number(readFileSync(pidFile, 'utf8')); @@ -394,7 +394,7 @@ enabled = true const missingManifest = path.join(tempRoot, 'missing', 'relay-plugin.toml'); await assert.rejects( () => - plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), activationSpec('missing_native', 'rust_dynamic', missingManifest), ]), @@ -408,13 +408,13 @@ enabled = true await assert.rejects( () => - plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef, { reject: true }), ]), /fixture rejection requested/i, ); - const recovered = await plugin.activateDynamicPlugins({ version: 1, components: [] }, [ + const recovered = await plugin.initializeWithDynamicPlugins({ version: 1, components: [] }, [ activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef), ]); await recovered.close(); @@ -428,7 +428,7 @@ enabled = true const plugin = require(${JSON.stringify(pluginModule)}); const config = { version: 1, components: [] }; const specs = [${JSON.stringify(activationSpec('fixture_native', 'rust_dynamic', nativeManifestRef))}]; - let activation = await plugin.activateDynamicPlugins(config, specs); + let activation = await plugin.initializeWithDynamicPlugins(config, specs); const weak = new WeakRef(activation); activation = null; let collected = false; @@ -450,7 +450,7 @@ enabled = true global.gc(); await new Promise((resolve) => setImmediate(resolve)); try { - replacement = await plugin.activateDynamicPlugins(config, specs); + replacement = await plugin.initializeWithDynamicPlugins(config, specs); break; } catch (error) { lastError = error; @@ -485,7 +485,7 @@ enabled = true const plugin = require(${JSON.stringify(pluginModule)}); const config = { version: 1, components: [] }; const specs = [${JSON.stringify(activationSpec('fixture_worker', 'worker', manifestRef))}]; - let activation = await plugin.activateDynamicPlugins(config, specs); + let activation = await plugin.initializeWithDynamicPlugins(config, specs); const weak = new WeakRef(activation); activation = null; await new Promise((resolve) => setImmediate(resolve)); @@ -522,7 +522,7 @@ enabled = true for (let index = 0; index < 500; index += 1) { await new Promise((resolve) => setTimeout(resolve, 10)); try { - replacement = await plugin.activateDynamicPlugins(config, specs); + replacement = await plugin.initializeWithDynamicPlugins(config, specs); break; } catch (error) { lastError = error;