diff --git a/Cargo.lock b/Cargo.lock index c1e6dadaa..3a2fd4dbc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1581,6 +1581,7 @@ dependencies = [ "nemo-relay-pii-redaction", "serde", "serde_json", + "tempfile", "tokio", "tokio-stream", "uuid", diff --git a/crates/ffi/Cargo.toml b/crates/ffi/Cargo.toml index f5628dfad..aad8a3c20 100644 --- a/crates/ffi/Cargo.toml +++ b/crates/ffi/Cargo.toml @@ -17,7 +17,7 @@ workspace = true crate-type = ["cdylib", "staticlib", "rlib"] [dependencies] -nemo-relay = { workspace = true, features = ["atof-streaming", "otel", "openinference"] } +nemo-relay = { workspace = true, features = ["atof-streaming", "otel", "openinference", "worker-grpc"] } nemo-relay-adaptive = { workspace = true, features = ["redis-backend"] } nemo-relay-pii-redaction.workspace = true chrono = "0.4" @@ -30,3 +30,6 @@ tokio-stream = "0.1" [build-dependencies] cbindgen = "0.29.2" + +[dev-dependencies] +tempfile = "3" diff --git a/crates/ffi/README.md b/crates/ffi/README.md index 3b70e7f15..ceb5e5cad 100644 --- a/crates/ffi/README.md +++ b/crates/ffi/README.md @@ -23,6 +23,11 @@ shared Rust runtime contract. This surface is experimental and source-first. The repository-maintained Go binding consumes it through CGo. +> **DO NOT TREAT AS PRODUCTION-READY:** the experimental +> `nemo_relay_initialize_with_dynamic_plugins` lifecycle needs a real consumer +> to validate shutdown, ownership, and error handling before it can be promoted +> to a stable contract. + ## Why Use It? - **Expose NeMo Relay to native consumers**: Call the shared Rust runtime from diff --git a/crates/ffi/nemo_relay.h b/crates/ffi/nemo_relay.h index 6355d579c..90bcdf566 100644 --- a/crates/ffi/nemo_relay.h +++ b/crates/ffi/nemo_relay.h @@ -159,6 +159,14 @@ typedef struct FfiOpenInferenceSubscriber FfiOpenInferenceSubscriber; */ typedef struct FfiOpenTelemetrySubscriber FfiOpenTelemetrySubscriber; +/** + * Opaque owned dynamic plugin host activation. + * + * The inner option allows explicit activation cleanup to be idempotent while + * retaining a stable allocation until the foreign caller frees the handle. + */ +typedef struct FfiPluginActivation FfiPluginActivation; + /** * Opaque plugin registration context. * @@ -1408,6 +1416,72 @@ NemoRelayStatus nemo_relay_openinference_subscriber_force_flush(const struct Ffi */ NemoRelayStatus nemo_relay_openinference_subscriber_shutdown(const struct FfiOpenInferenceSubscriber *subscriber); +/** + * Load and activate dynamic plugins as one owned transaction. + * + * **Experimental:** this API needs a production consumer before its lifecycle + * contract is considered stable. + * + * Relay discovers `plugins.toml` once during this startup call and layers + * `config_json` over the discovered configuration. Explicit values take + * precedence where both sources configure the same setting. Static components + * from the resolved configuration initialize before components appended by + * the dynamic plugins. Configuration files are not watched or reloaded. + * `dynamic_plugins_json` must contain at least one explicit dynamic-plugin + * activation specification; use `nemo_relay_initialize_plugins` for a + * static-only configuration. + * + * The explicit configuration uses this JSON shape: + * + * ```text + * {"version":1,"components":[{"kind":"static.kind","enabled":true,"config":{}}]} + * ``` + * + * Dynamic plugin specifications use this JSON shape: + * + * ```text + * [{"plugin_id":"example","kind":"rust_dynamic","manifest_ref":"/absolute/path/relay-plugin.toml","config":{}}] + * ``` + * + * `kind` must be `rust_dynamic` or `worker`. `environment_ref` is optional + * and applies only to worker plugins. `manifest_ref` is resolved by the + * embedding application; this API does not discover installed plugins. + * + * On success, the caller owns `out_activation` and + * must clear and free it with `nemo_relay_plugin_activation_clear` and + * `nemo_relay_plugin_activation_free`. `out_report_json` is a library-owned C + * string and must be released with `nemo_relay_string_free`. + * + * # Safety + * Both input pointers must reference valid, null-terminated C strings. + * `out_activation` and `out_report_json` must be valid, non-null, non-overlapping + * output pointers. + */ +NemoRelayStatus nemo_relay_initialize_with_dynamic_plugins(const char *config_json, + const char *dynamic_plugins_json, + struct FfiPluginActivation **out_activation, + char **out_report_json); + +/** + * Clear one owned dynamic plugin activation. + * + * This operation is idempotent. A null handle is treated as already cleared. + * If teardown fails, the error is reported only by the call that performs the + * teardown. The activation is consumed regardless of the outcome, so a later + * clear returns success and does not report the earlier error again. + * Concurrent clear calls for the same handle are serialized, but they must not + * overlap with `nemo_relay_plugin_activation_free`. + * The handle allocation remains owned by the caller and must still be passed + * to `nemo_relay_plugin_activation_free`. + * + * # Safety + * `activation` must be a valid activation handle returned by + * `nemo_relay_initialize_with_dynamic_plugins`, or null. The caller must ensure the + * handle remains allocated for this call and that + * `nemo_relay_plugin_activation_free` does not run concurrently with it. + */ +NemoRelayStatus nemo_relay_plugin_activation_clear(struct FfiPluginActivation *activation); + /** * Validate a generic plugin config document and return the diagnostics report as JSON. * @@ -2451,6 +2525,25 @@ void nemo_relay_openinference_subscriber_free(struct FfiOpenInferenceSubscriber */ void nemo_relay_adaptive_runtime_free(struct FfiAdaptiveRuntime *ptr); +/** + * Free a dynamic plugin activation handle previously returned by + * `nemo_relay_initialize_with_dynamic_plugins`. + * + * Any activation that has not already been explicitly cleared is cleaned up + * best-effort by its Rust destructor before the allocation is released. The + * caller's handle is set to null before cleanup, making repeated calls through + * the same handle variable safe when they are sequential. + * + * # Safety + * `ptr` must be null or point to a writable activation-handle variable whose + * value is null or was returned by `nemo_relay_initialize_with_dynamic_plugins`. The + * caller must ensure that no operation, including + * `nemo_relay_plugin_activation_clear`, accesses the activation concurrently + * with this call and that no operation can use the handle after this call + * begins. + */ +void nemo_relay_plugin_activation_free(struct FfiPluginActivation **ptr); + /** * Free a codec handle previously returned by one of the codec constructor * functions (`nemo_relay_openai_chat_codec_new`, etc.). diff --git a/crates/ffi/src/api/mod.rs b/crates/ffi/src/api/mod.rs index 1d62d9e69..27817a82b 100644 --- a/crates/ffi/src/api/mod.rs +++ b/crates/ffi/src/api/mod.rs @@ -36,8 +36,8 @@ use crate::error::{ }; use crate::types::{ FfiAtifExporter, FfiAtofExporter, FfiCodecHandle, FfiLLMHandle, FfiLLMRequest, - FfiOpenInferenceSubscriber, FfiOpenTelemetrySubscriber, FfiPluginContext, FfiScopeHandle, - FfiScopeStack, FfiThreadScopeStackBinding, FfiToolHandle, NemoRelayScopeType, + FfiOpenInferenceSubscriber, FfiOpenTelemetrySubscriber, FfiPluginActivation, FfiPluginContext, + FfiScopeHandle, FfiScopeStack, FfiThreadScopeStackBinding, FfiToolHandle, NemoRelayScopeType, }; pub use crate::types::{nemo_relay_openinference_subscriber_free, nemo_relay_otel_subscriber_free}; use libc::c_char; @@ -57,6 +57,7 @@ use nemo_relay::api::tool as core_tool_api; use nemo_relay::api::tool::ToolAttributes; use nemo_relay::codec::optimization::LlmOptimizationContribution; use nemo_relay::error::Result as FlowResult; +use nemo_relay::plugin::dynamic::{DynamicPluginActivationSpec, PluginHostActivation}; use nemo_relay::plugin::{ ConfigDiagnostic, DiagnosticLevel, Plugin, PluginConfig, PluginError, PluginRegistrationContext, active_plugin_report, clear_plugin_configuration, deregister_plugin, diff --git a/crates/ffi/src/api/plugin.rs b/crates/ffi/src/api/plugin.rs index 51021e09e..9130a2f94 100644 --- a/crates/ffi/src/api/plugin.rs +++ b/crates/ffi/src/api/plugin.rs @@ -2,20 +2,21 @@ // SPDX-License-Identifier: Apache-2.0 use super::{ - Arc, CStr, ConfigDiagnostic, DiagnosticLevel, FfiPluginContext, Future, - NemoRelayEventSanitizeCb, NemoRelayEventSubscriberCb, NemoRelayFreeFn, NemoRelayJsonCb, - NemoRelayLlmConditionalCb, NemoRelayLlmExecInterceptCb, NemoRelayLlmRequestCb, - NemoRelayLlmRequestInterceptCb, NemoRelayPluginRegisterCb, NemoRelayPluginValidateCb, - NemoRelayStatus, NemoRelayToolConditionalCb, NemoRelayToolExecInterceptCb, - NemoRelayToolSanitizeCb, Pin, Plugin, PluginConfig, PluginError, PluginRegistrationContext, - active_plugin_report, c_char, c_str_to_json, c_str_to_string, clear_last_error, - clear_plugin_configuration, deregister_plugin, initialize_plugins, json_to_c_string, - last_error_message, list_plugin_kinds, nemo_relay_string_free, register_adaptive_component, - register_plugin, set_last_error, status_from_plugin_error, tokio_runtime, - validate_plugin_config, wrap_event_sanitize_fn, wrap_event_subscriber, wrap_llm_conditional_fn, - wrap_llm_exec_intercept_fn, wrap_llm_request_intercept_fn, wrap_llm_response_fn, - wrap_llm_sanitize_request_fn, wrap_llm_stream_exec_intercept_fn, wrap_tool_conditional_fn, - wrap_tool_exec_intercept_fn, wrap_tool_request_intercept_fn, wrap_tool_sanitize_fn, + Arc, CStr, ConfigDiagnostic, DiagnosticLevel, DynamicPluginActivationSpec, FfiPluginActivation, + FfiPluginContext, Future, NemoRelayEventSanitizeCb, NemoRelayEventSubscriberCb, + NemoRelayFreeFn, NemoRelayJsonCb, NemoRelayLlmConditionalCb, NemoRelayLlmExecInterceptCb, + NemoRelayLlmRequestCb, NemoRelayLlmRequestInterceptCb, NemoRelayPluginRegisterCb, + NemoRelayPluginValidateCb, NemoRelayStatus, NemoRelayToolConditionalCb, + NemoRelayToolExecInterceptCb, NemoRelayToolSanitizeCb, Pin, Plugin, PluginConfig, PluginError, + PluginHostActivation, PluginRegistrationContext, active_plugin_report, c_char, c_str_to_json, + c_str_to_string, clear_last_error, clear_plugin_configuration, deregister_plugin, + initialize_plugins, json_to_c_string, last_error_message, list_plugin_kinds, + nemo_relay_string_free, register_adaptive_component, register_plugin, set_last_error, + status_from_plugin_error, tokio_runtime, validate_plugin_config, wrap_event_sanitize_fn, + wrap_event_subscriber, wrap_llm_conditional_fn, wrap_llm_exec_intercept_fn, + wrap_llm_request_intercept_fn, wrap_llm_response_fn, wrap_llm_sanitize_request_fn, + wrap_llm_stream_exec_intercept_fn, wrap_tool_conditional_fn, wrap_tool_exec_intercept_fn, + wrap_tool_request_intercept_fn, wrap_tool_sanitize_fn, }; use crate::api::event_registry::Surface; use nemo_relay_pii_redaction::component::register_pii_redaction_component; @@ -132,6 +133,180 @@ fn ensure_pii_redaction_component_registered() -> std::result::Result<(), NemoRe register_pii_redaction_component().map_err(|err| status_from_plugin_error(&err)) } +fn parse_plugin_config( + config_json: *const c_char, +) -> std::result::Result { + let value = c_str_to_json(config_json).ok_or(NemoRelayStatus::InvalidJson)?; + serde_json::from_value(value).map_err(|error| { + set_last_error(&format!("invalid plugin config: {error}")); + NemoRelayStatus::InvalidJson + }) +} + +fn parse_dynamic_plugin_specs( + dynamic_plugins_json: *const c_char, +) -> std::result::Result, NemoRelayStatus> { + let value = c_str_to_json(dynamic_plugins_json).ok_or(NemoRelayStatus::InvalidJson)?; + serde_json::from_value(value).map_err(|error| { + set_last_error(&format!("invalid dynamic plugin specifications: {error}")); + NemoRelayStatus::InvalidJson + }) +} + +fn lock_plugin_activation( + activation: &FfiPluginActivation, +) -> std::result::Result>, NemoRelayStatus> { + activation.0.lock().map_err(|error| { + set_last_error(&format!("plugin activation lock poisoned: {error}")); + NemoRelayStatus::Internal + }) +} + +/// Load and activate dynamic plugins as one owned transaction. +/// +/// **Experimental:** this API needs a production consumer before its lifecycle +/// contract is considered stable. +/// +/// Relay discovers `plugins.toml` once during this startup call and layers +/// `config_json` over the discovered configuration. Explicit values take +/// precedence where both sources configure the same setting. Static components +/// from the resolved configuration initialize before components appended by +/// the dynamic plugins. Configuration files are not watched or reloaded. +/// `dynamic_plugins_json` must contain at least one explicit dynamic-plugin +/// activation specification; use `nemo_relay_initialize_plugins` for a +/// static-only configuration. +/// +/// The explicit configuration uses this JSON shape: +/// +/// ```text +/// {"version":1,"components":[{"kind":"static.kind","enabled":true,"config":{}}]} +/// ``` +/// +/// Dynamic plugin specifications use this JSON shape: +/// +/// ```text +/// [{"plugin_id":"example","kind":"rust_dynamic","manifest_ref":"/absolute/path/relay-plugin.toml","config":{}}] +/// ``` +/// +/// `kind` must be `rust_dynamic` or `worker`. `environment_ref` is optional +/// and applies only to worker plugins. `manifest_ref` is resolved by the +/// embedding application; this API does not discover installed plugins. +/// +/// On success, the caller owns `out_activation` and +/// must clear and free it with `nemo_relay_plugin_activation_clear` and +/// `nemo_relay_plugin_activation_free`. `out_report_json` is a library-owned C +/// string and must be released with `nemo_relay_string_free`. +/// +/// # Safety +/// Both input pointers must reference valid, null-terminated C strings. +/// `out_activation` and `out_report_json` must be valid, non-null, non-overlapping +/// output pointers. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_relay_initialize_with_dynamic_plugins( + config_json: *const c_char, + dynamic_plugins_json: *const c_char, + out_activation: *mut *mut FfiPluginActivation, + out_report_json: *mut *mut c_char, +) -> NemoRelayStatus { + clear_last_error(); + if out_activation.is_null() { + if !out_report_json.is_null() { + unsafe { *out_report_json = std::ptr::null_mut() }; + } + set_last_error("out_activation pointer is null"); + return NemoRelayStatus::NullPointer; + } + if out_report_json.is_null() { + unsafe { *out_activation = std::ptr::null_mut() }; + set_last_error("out_report_json pointer is null"); + return NemoRelayStatus::NullPointer; + } + if out_activation.cast::() == out_report_json.cast::() { + unsafe { *out_activation = std::ptr::null_mut() }; + set_last_error("out_activation and out_report_json must not overlap"); + return NemoRelayStatus::InvalidArg; + } + unsafe { + *out_activation = std::ptr::null_mut(); + *out_report_json = std::ptr::null_mut(); + } + + if let Err(status) = ensure_adaptive_component_registered() { + return status; + } + if let Err(status) = ensure_pii_redaction_component_registered() { + return status; + } + let config = match parse_plugin_config(config_json) { + Ok(config) => config, + Err(status) => return status, + }; + let dynamic_plugins = match parse_dynamic_plugin_specs(dynamic_plugins_json) { + Ok(dynamic_plugins) => dynamic_plugins, + Err(status) => return status, + }; + let (activation, report) = match tokio_runtime().block_on( + PluginHostActivation::activate_with_discovered_config(config, dynamic_plugins), + ) { + Ok(result) => result, + Err(error) => return status_from_plugin_error(&error), + }; + let report_json = match serde_json::to_value(report) { + Ok(value) => value, + Err(error) => { + let _ = activation.clear(); + set_last_error(&error.to_string()); + return NemoRelayStatus::Internal; + } + }; + + unsafe { + *out_activation = Box::into_raw(Box::new(FfiPluginActivation(std::sync::Mutex::new( + Some(activation), + )))); + *out_report_json = json_to_c_string(&report_json); + } + NemoRelayStatus::Ok +} + +/// Clear one owned dynamic plugin activation. +/// +/// This operation is idempotent. A null handle is treated as already cleared. +/// If teardown fails, the error is reported only by the call that performs the +/// teardown. The activation is consumed regardless of the outcome, so a later +/// clear returns success and does not report the earlier error again. +/// Concurrent clear calls for the same handle are serialized, but they must not +/// overlap with `nemo_relay_plugin_activation_free`. +/// The handle allocation remains owned by the caller and must still be passed +/// to `nemo_relay_plugin_activation_free`. +/// +/// # Safety +/// `activation` must be a valid activation handle returned by +/// `nemo_relay_initialize_with_dynamic_plugins`, or null. The caller must ensure the +/// handle remains allocated for this call and that +/// `nemo_relay_plugin_activation_free` does not run concurrently with it. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_relay_plugin_activation_clear( + activation: *mut FfiPluginActivation, +) -> NemoRelayStatus { + clear_last_error(); + if activation.is_null() { + return NemoRelayStatus::Ok; + } + let activation = unsafe { &*activation }; + let mut guard = match lock_plugin_activation(activation) { + Ok(guard) => guard, + Err(status) => return status, + }; + let Some(activation) = guard.take() else { + return NemoRelayStatus::Ok; + }; + match activation.clear() { + Ok(()) => NemoRelayStatus::Ok, + Err(error) => status_from_plugin_error(&error), + } +} + /// Validate a generic plugin config document and return the diagnostics report as JSON. /// /// # Safety diff --git a/crates/ffi/src/types/mod.rs b/crates/ffi/src/types/mod.rs index 20cf03cd1..e66f0c4e1 100644 --- a/crates/ffi/src/types/mod.rs +++ b/crates/ffi/src/types/mod.rs @@ -26,6 +26,7 @@ use nemo_relay::api::scope::{ScopeHandle, ScopeType}; use nemo_relay::api::tool::ToolAttributes; use nemo_relay::api::tool::ToolHandle; use nemo_relay::codec::traits::{LlmCodec, LlmResponseCodec}; +use nemo_relay::plugin::dynamic::PluginHostActivation; use nemo_relay_adaptive::AdaptiveRuntime; use crate::convert::{json_to_c_string, str_to_c_string}; @@ -64,6 +65,11 @@ pub struct FfiOpenInferenceSubscriber( ); /// Opaque owned adaptive runtime handle. pub struct FfiAdaptiveRuntime(pub std::sync::Mutex>); +/// Opaque owned dynamic plugin host activation. +/// +/// The inner option allows explicit activation cleanup to be idempotent while +/// retaining a stable allocation until the foreign caller frees the handle. +pub struct FfiPluginActivation(pub std::sync::Mutex>); /// Opaque plugin registration context. /// /// This wrapper contains a borrowed raw pointer to an @@ -287,6 +293,32 @@ pub unsafe extern "C" fn nemo_relay_adaptive_runtime_free(ptr: *mut FfiAdaptiveR } } +/// Free a dynamic plugin activation handle previously returned by +/// `nemo_relay_initialize_with_dynamic_plugins`. +/// +/// Any activation that has not already been explicitly cleared is cleaned up +/// best-effort by its Rust destructor before the allocation is released. The +/// caller's handle is set to null before cleanup, making repeated calls through +/// the same handle variable safe when they are sequential. +/// +/// # Safety +/// `ptr` must be null or point to a writable activation-handle variable whose +/// value is null or was returned by `nemo_relay_initialize_with_dynamic_plugins`. The +/// caller must ensure that no operation, including +/// `nemo_relay_plugin_activation_clear`, accesses the activation concurrently +/// with this call and that no operation can use the handle after this call +/// begins. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_relay_plugin_activation_free(ptr: *mut *mut FfiPluginActivation) { + if ptr.is_null() { + return; + } + let activation = unsafe { ptr.replace(std::ptr::null_mut()) }; + if !activation.is_null() { + drop(unsafe { Box::from_raw(activation) }); + } +} + /// Free a codec handle previously returned by one of the codec constructor /// functions (`nemo_relay_openai_chat_codec_new`, etc.). /// diff --git a/crates/ffi/tests/integration/api_tests.rs b/crates/ffi/tests/integration/api_tests.rs index 8d9e11b8d..5afdadf8c 100644 --- a/crates/ffi/tests/integration/api_tests.rs +++ b/crates/ffi/tests/integration/api_tests.rs @@ -7,7 +7,7 @@ use super::*; use std::ffi::{CStr, CString}; use std::fs; use std::ptr; -use std::sync::{Mutex, OnceLock}; +use std::sync::OnceLock; use std::time::{SystemTime, UNIX_EPOCH}; use nemo_relay::plugin::PluginRegistrationContext; @@ -19,24 +19,24 @@ use nemo_relay_ffi::convert::nemo_relay_string_free; use nemo_relay_ffi::error::{NemoRelayStatus, nemo_relay_last_error, set_last_error}; use nemo_relay_ffi::types::{ FfiAtifExporter, FfiAtofExporter, FfiEvent, FfiLLMHandle, FfiLLMRequest, - FfiOpenTelemetrySubscriber, FfiScopeStack, FfiToolHandle, nemo_relay_atif_exporter_free, - nemo_relay_atof_exporter_free, nemo_relay_event_data, nemo_relay_event_input, - nemo_relay_event_metadata, nemo_relay_event_model_name, nemo_relay_event_name, - nemo_relay_event_output, nemo_relay_event_parent_uuid, nemo_relay_event_scope_type, - nemo_relay_event_timestamp, nemo_relay_event_tool_call_id, nemo_relay_event_uuid, - nemo_relay_llm_handle_attributes, nemo_relay_llm_handle_free, nemo_relay_llm_handle_name, - nemo_relay_llm_handle_parent_uuid, nemo_relay_llm_handle_uuid, nemo_relay_llm_request_content, - nemo_relay_llm_request_free, nemo_relay_llm_request_headers, nemo_relay_llm_request_new, - nemo_relay_otel_subscriber_free, nemo_relay_scope_handle_attributes, - nemo_relay_scope_handle_data, nemo_relay_scope_handle_free, nemo_relay_scope_handle_metadata, - nemo_relay_scope_handle_name, nemo_relay_scope_handle_parent_uuid, - nemo_relay_scope_handle_scope_type, nemo_relay_scope_handle_uuid, nemo_relay_scope_stack_free, - nemo_relay_tool_handle_attributes, nemo_relay_tool_handle_free, nemo_relay_tool_handle_name, - nemo_relay_tool_handle_parent_uuid, nemo_relay_tool_handle_uuid, + FfiOpenTelemetrySubscriber, FfiPluginActivation, FfiScopeStack, FfiToolHandle, + nemo_relay_atif_exporter_free, nemo_relay_atof_exporter_free, nemo_relay_event_data, + nemo_relay_event_input, nemo_relay_event_metadata, nemo_relay_event_model_name, + nemo_relay_event_name, nemo_relay_event_output, nemo_relay_event_parent_uuid, + nemo_relay_event_scope_type, nemo_relay_event_timestamp, nemo_relay_event_tool_call_id, + nemo_relay_event_uuid, nemo_relay_llm_handle_attributes, nemo_relay_llm_handle_free, + nemo_relay_llm_handle_name, nemo_relay_llm_handle_parent_uuid, nemo_relay_llm_handle_uuid, + nemo_relay_llm_request_content, nemo_relay_llm_request_free, nemo_relay_llm_request_headers, + nemo_relay_llm_request_new, nemo_relay_otel_subscriber_free, + nemo_relay_scope_handle_attributes, nemo_relay_scope_handle_data, nemo_relay_scope_handle_free, + nemo_relay_scope_handle_metadata, nemo_relay_scope_handle_name, + nemo_relay_scope_handle_parent_uuid, nemo_relay_scope_handle_scope_type, + nemo_relay_scope_handle_uuid, nemo_relay_scope_stack_free, nemo_relay_tool_handle_attributes, + nemo_relay_tool_handle_free, nemo_relay_tool_handle_name, nemo_relay_tool_handle_parent_uuid, + nemo_relay_tool_handle_uuid, }; use nemo_relay_ffi::{api, callable, types}; -static TEST_MUTEX: Mutex<()> = Mutex::new(()); static EVENT_LOG: OnceLock>> = OnceLock::new(); static COLLECTED_CHUNKS: OnceLock>> = OnceLock::new(); static FINALIZER_CALLS: OnceLock> = OnceLock::new(); diff --git a/crates/ffi/tests/integration/main.rs b/crates/ffi/tests/integration/main.rs index 04e80d2d4..1aacee267 100644 --- a/crates/ffi/tests/integration/main.rs +++ b/crates/ffi/tests/integration/main.rs @@ -20,9 +20,11 @@ use nemo_relay_ffi::{api, convert, error}; use serde_json::{Value as Json, json}; use std::ffi::{CStr, CString}; use std::pin::Pin; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use tokio_stream::Stream; +static TEST_MUTEX: Mutex<()> = Mutex::new(()); + unsafe fn nemo_relay_string_free_internal(ptr: *mut c_char) { unsafe { nemo_relay_string_free(ptr) }; } @@ -35,5 +37,6 @@ mod callable_tests; mod convert_coverage_tests; #[path = "../coverage/error_tests.rs"] mod error_coverage_tests; +mod plugin_activation_tests; #[path = "../unit/types_tests.rs"] mod types_tests; diff --git a/crates/ffi/tests/integration/plugin_activation_tests.rs b/crates/ffi/tests/integration/plugin_activation_tests.rs new file mode 100644 index 000000000..5ecfb74a6 --- /dev/null +++ b/crates/ffi/tests/integration/plugin_activation_tests.rs @@ -0,0 +1,652 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::ptr; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Mutex, OnceLock}; + +use nemo_relay_ffi::types::{FfiPluginActivation, nemo_relay_plugin_activation_free}; +use tempfile::TempDir; + +const DISCOVERY_CHILD_ENV: &str = "NEMO_RELAY_FFI_DISCOVERY_CHILD"; +const DISCOVERED_STATIC_PLUGIN_KIND: &str = "ffi_discovered_static"; +static DISCOVERED_STATIC_REGISTRATIONS: AtomicUsize = AtomicUsize::new(0); +static DISCOVERED_STATIC_CALLBACKS: AtomicUsize = AtomicUsize::new(0); +static DISCOVERED_STATIC_CONFIG: Mutex> = Mutex::new(None); + +struct PluginDiscoveryTestEnv { + previous_cwd: PathBuf, + previous_xdg_config_home: Option, +} + +impl PluginDiscoveryTestEnv { + fn enter(cwd: &Path, xdg_config_home: &Path) -> Self { + let guard = Self { + previous_cwd: std::env::current_dir().expect("current directory"), + previous_xdg_config_home: std::env::var_os("XDG_CONFIG_HOME"), + }; + std::env::set_current_dir(cwd).expect("set project directory"); + // SAFETY: this runs in a dedicated child test process and Drop restores + // the environment before that process exits. + unsafe { std::env::set_var("XDG_CONFIG_HOME", xdg_config_home) }; + guard + } +} + +impl Drop for PluginDiscoveryTestEnv { + fn drop(&mut self) { + let _ = std::env::set_current_dir(&self.previous_cwd); + // SAFETY: see PluginDiscoveryTestEnv::enter. + unsafe { + match &self.previous_xdg_config_home { + Some(value) => std::env::set_var("XDG_CONFIG_HOME", value), + None => std::env::remove_var("XDG_CONFIG_HOME"), + } + } + } +} + +#[test] +fn ffi_activation_layers_discovered_static_and_explicit_dynamic_plugins() { + if std::env::var_os(DISCOVERY_CHILD_ENV).is_some() { + run_discovered_config_activation_test(); + return; + } + + let output = Command::new(std::env::current_exe().expect("current test executable")) + .arg("--exact") + .arg( + "plugin_activation_tests::ffi_activation_layers_discovered_static_and_explicit_dynamic_plugins", + ) + .arg("--nocapture") + .env(DISCOVERY_CHILD_ENV, "1") + .output() + .expect("discovery child test should start"); + assert!( + output.status.success(), + "discovery child test failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +fn run_discovered_config_activation_test() { + let _ = nemo_relay_clear_plugin_configuration(); + DISCOVERED_STATIC_REGISTRATIONS.store(0, Ordering::SeqCst); + DISCOVERED_STATIC_CALLBACKS.store(0, Ordering::SeqCst); + *DISCOVERED_STATIC_CONFIG.lock().unwrap() = None; + + let environment = TempDir::new().expect("plugin discovery environment"); + let project_config_dir = environment.path().join(".nemo-relay"); + let xdg_config_home = environment.path().join("xdg"); + std::fs::create_dir_all(&project_config_dir).expect("project config directory"); + std::fs::create_dir_all(&xdg_config_home).expect("isolated user config directory"); + let plugins_toml = project_config_dir.join("plugins.toml"); + std::fs::write(&plugins_toml, "invalid = [").expect("write invalid plugin config"); + let _environment = PluginDiscoveryTestEnv::enter(environment.path(), &xdg_config_home); + + // Empty specifications fail before discovery or ownership. The malformed + // file would otherwise produce a TOML error, and the successful activation + // below proves this attempt did not retain the process-wide host claim. + let config = cstring(r#"{"version":1,"components":[]}"#); + let empty_specs = cstring("[]"); + let mut empty_activation = ptr::null_mut(); + let mut empty_report = ptr::null_mut(); + assert_eq!( + unsafe { + api::nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + empty_specs.as_ptr(), + &mut empty_activation, + &mut empty_report, + ) + }, + NemoRelayStatus::InvalidArg + ); + assert!(empty_activation.is_null()); + assert!(empty_report.is_null()); + assert!( + unsafe { read_last_error() } + .unwrap_or_default() + .contains("at least one dynamic plugin") + ); + + std::fs::write( + &plugins_toml, + format!( + r#"version = 999 + +[[components]] +kind = {DISCOVERED_STATIC_PLUGIN_KIND:?} +enabled = true + +[components.config] +source = "project-file" +"# + ), + ) + .expect("write project plugin config"); + + let plugin_kind = cstring(DISCOVERED_STATIC_PLUGIN_KIND); + assert_eq!( + unsafe { + api::nemo_relay_register_plugin( + plugin_kind.as_ptr(), + None, + discovered_static_register, + ptr::null_mut(), + None, + ) + }, + NemoRelayStatus::Ok + ); + + let manifest_dir = TempDir::new().expect("native manifest tempdir"); + let manifest = write_native_manifest(manifest_dir.path(), build_native_fixture()); + let (mut activation, report) = initialize_with_dynamic_plugins(json!([{ + "plugin_id": "fixture_native", + "kind": "rust_dynamic", + "manifest_ref": manifest, + "config": {} + }])); + + // The explicit version 1 must override the discovered version 999. The + // file-only component and its config must still survive the merge. + assert_eq!(report["diagnostics"], json!([])); + assert_eq!(DISCOVERED_STATIC_REGISTRATIONS.load(Ordering::SeqCst), 1); + assert_eq!( + DISCOVERED_STATIC_CONFIG.lock().unwrap().as_ref(), + Some(&json!({"source": "project-file"})) + ); + assert!(plugin_kinds().iter().any(|kind| kind == "fixture_native")); + + // Mutating the file after startup has no effect: discovery is one-shot. + std::fs::write(&plugins_toml, "invalid = [").expect("mutate plugin config after startup"); + let intercepted = tool_request_intercepts("ffi-layered-tool", json!({"input": true})); + assert_eq!(intercepted["file_static"], true); + assert_eq!(intercepted["static_saw_dynamic"], false); + assert_eq!(intercepted["native_plugin"], true); + assert_eq!(DISCOVERED_STATIC_CALLBACKS.load(Ordering::SeqCst), 1); + + unsafe { + assert_eq!( + api::nemo_relay_plugin_activation_clear(activation), + NemoRelayStatus::Ok + ); + nemo_relay_plugin_activation_free(&mut activation); + } + assert!(!plugin_kinds().iter().any(|kind| kind == "fixture_native")); + assert_eq!( + tool_request_intercepts("ffi-layered-tool", json!({"input": true})), + json!({"input": true}) + ); + assert_eq!(DISCOVERED_STATIC_CALLBACKS.load(Ordering::SeqCst), 1); + assert_eq!( + unsafe { api::nemo_relay_deregister_plugin(plugin_kind.as_ptr()) }, + NemoRelayStatus::Ok + ); +} + +unsafe extern "C" fn discovered_static_register( + _user_data: *mut libc::c_void, + plugin_config_json: *const c_char, + ctx: *mut FfiPluginContext, +) -> NemoRelayStatus { + let config = unsafe { CStr::from_ptr(plugin_config_json) } + .to_str() + .ok() + .and_then(|value| serde_json::from_str(value).ok()); + *DISCOVERED_STATIC_CONFIG.lock().unwrap() = config; + DISCOVERED_STATIC_REGISTRATIONS.fetch_add(1, Ordering::SeqCst); + let name = cstring("project_file_intercept"); + unsafe { + api::nemo_relay_plugin_context_register_tool_request_intercept( + ctx, + name.as_ptr(), + -1, + false, + discovered_static_tool_request, + ptr::null_mut(), + None, + ) + } +} + +unsafe extern "C" fn discovered_static_tool_request( + _user_data: *mut libc::c_void, + _name: *const c_char, + args_json: *const c_char, +) -> *mut c_char { + DISCOVERED_STATIC_CALLBACKS.fetch_add(1, Ordering::SeqCst); + let mut args: Json = serde_json::from_str( + unsafe { CStr::from_ptr(args_json) } + .to_str() + .unwrap_or("null"), + ) + .unwrap_or_else(|_| json!({})); + args["static_saw_dynamic"] = json!(args.get("native_plugin").is_some()); + args["file_static"] = json!(true); + CString::new(args.to_string()).unwrap().into_raw() +} + +#[test] +fn ffi_activation_loads_native_callbacks_and_removes_them_before_free() { + let _guard = TEST_MUTEX.lock().unwrap(); + let _ = nemo_relay_clear_plugin_configuration(); + + let manifest_dir = TempDir::new().expect("native manifest tempdir"); + let manifest = write_native_manifest(manifest_dir.path(), build_native_fixture()); + let (mut activation, report) = initialize_with_dynamic_plugins(json!([{ + "plugin_id": "fixture_native", + "kind": "rust_dynamic", + "manifest_ref": manifest, + "config": {} + }])); + assert_eq!(report["diagnostics"], json!([])); + assert!(plugin_kinds().iter().any(|kind| kind == "fixture_native")); + + assert_eq!( + tool_request_intercepts("ffi-native-tool", json!({"input": true}))["native_plugin"], + true + ); + + unsafe { + assert_eq!( + api::nemo_relay_plugin_activation_clear(activation), + NemoRelayStatus::Ok + ); + assert_eq!( + api::nemo_relay_plugin_activation_clear(activation), + NemoRelayStatus::Ok + ); + nemo_relay_plugin_activation_free(&mut activation); + } + assert!(!plugin_kinds().iter().any(|kind| kind == "fixture_native")); + assert_eq!( + tool_request_intercepts("ffi-native-tool", json!({"input": true})), + json!({"input": true}) + ); + + let (mut drop_activation, _) = initialize_with_dynamic_plugins(json!([{ + "plugin_id": "fixture_native", + "kind": "rust_dynamic", + "manifest_ref": manifest, + "config": {} + }])); + assert_eq!( + tool_request_intercepts("ffi-native-tool", json!({"input": true}))["native_plugin"], + true + ); + unsafe { nemo_relay_plugin_activation_free(&mut drop_activation) }; + assert_eq!( + tool_request_intercepts("ffi-native-tool", json!({"input": true})), + json!({"input": true}) + ); +} + +#[test] +fn ffi_activation_rejects_overlapping_outputs_without_claiming_host() { + let _guard = TEST_MUTEX.lock().unwrap(); + let _ = nemo_relay_clear_plugin_configuration(); + + let manifest_dir = TempDir::new().expect("native manifest tempdir"); + let manifest = write_native_manifest(manifest_dir.path(), build_native_fixture()); + let config = cstring(r#"{"version":1,"components":[]}"#); + let specs_value = json!([{ + "plugin_id": "fixture_native", + "kind": "rust_dynamic", + "manifest_ref": manifest, + "config": {} + }]); + let specs = cstring(&specs_value.to_string()); + let mut aliased_output = std::ptr::dangling_mut::(); + let output_slot = &mut aliased_output as *mut *mut std::ffi::c_void; + let status = unsafe { + api::nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + output_slot.cast::<*mut FfiPluginActivation>(), + output_slot.cast::<*mut c_char>(), + ) + }; + assert_eq!(status, NemoRelayStatus::InvalidArg); + assert!(aliased_output.is_null()); + assert!( + unsafe { read_last_error() } + .unwrap_or_default() + .contains("must not overlap") + ); + + let (mut activation, _) = initialize_with_dynamic_plugins(specs_value); + unsafe { + assert_eq!( + api::nemo_relay_plugin_activation_clear(activation), + NemoRelayStatus::Ok + ); + nemo_relay_plugin_activation_free(&mut activation); + } +} + +#[test] +fn ffi_activation_loads_worker_callbacks_and_stops_worker_on_clear() { + let _guard = TEST_MUTEX.lock().unwrap(); + let _ = nemo_relay_clear_plugin_configuration(); + + let manifest_dir = TempDir::new().expect("worker manifest tempdir"); + let manifest = write_worker_manifest(manifest_dir.path(), build_worker_fixture()); + let (mut activation, report) = initialize_with_dynamic_plugins(json!([{ + "plugin_id": "fixture_worker", + "kind": "worker", + "manifest_ref": manifest, + "config": {} + }])); + assert_eq!(report["diagnostics"], json!([])); + assert!(plugin_kinds().iter().any(|kind| kind == "fixture_worker")); + assert_eq!( + tool_request_intercepts("ffi-worker-tool", json!({"input": true}))["worker_plugin"], + true + ); + + unsafe { + assert_eq!( + api::nemo_relay_plugin_activation_clear(activation), + NemoRelayStatus::Ok + ); + nemo_relay_plugin_activation_free(&mut activation); + } + assert!(!plugin_kinds().iter().any(|kind| kind == "fixture_worker")); + assert_eq!( + tool_request_intercepts("ffi-worker-tool", json!({"input": true})), + json!({"input": true}) + ); +} + +#[test] +fn ffi_activation_rolls_back_an_earlier_native_load_when_a_later_load_fails() { + let _guard = TEST_MUTEX.lock().unwrap(); + let _ = nemo_relay_clear_plugin_configuration(); + + let manifest_dir = TempDir::new().expect("native manifest tempdir"); + let manifest = write_native_manifest(manifest_dir.path(), build_native_fixture()); + let missing_manifest = manifest_dir.path().join("missing-relay-plugin.toml"); + let config = cstring(r#"{"version":1,"components":[]}"#); + let specs = cstring( + &json!([ + { + "plugin_id": "fixture_native", + "kind": "rust_dynamic", + "manifest_ref": manifest, + "config": {} + }, + { + "plugin_id": "fixture_missing", + "kind": "rust_dynamic", + "manifest_ref": missing_manifest, + "config": {} + } + ]) + .to_string(), + ); + let mut activation = ptr::null_mut(); + let mut report = ptr::null_mut(); + let status = unsafe { + api::nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + &mut activation, + &mut report, + ) + }; + assert_eq!(status, NemoRelayStatus::NotFound); + assert!(activation.is_null()); + assert!(report.is_null()); + assert!(!plugin_kinds().iter().any(|kind| kind == "fixture_native")); + assert_eq!( + tool_request_intercepts("ffi-native-tool", json!({"input": true})), + json!({"input": true}) + ); + + let (mut activation, _) = initialize_with_dynamic_plugins(json!([{ + "plugin_id": "fixture_native", + "kind": "rust_dynamic", + "manifest_ref": manifest, + "config": {} + }])); + unsafe { + assert_eq!( + api::nemo_relay_plugin_activation_clear(activation), + NemoRelayStatus::Ok + ); + nemo_relay_plugin_activation_free(&mut activation); + } +} + +fn initialize_with_dynamic_plugins(specs: Json) -> (*mut FfiPluginActivation, Json) { + let config = cstring(r#"{"version":1,"components":[]}"#); + let specs = cstring(&specs.to_string()); + let mut activation = ptr::null_mut(); + let mut report = ptr::null_mut(); + let status = unsafe { + api::nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + &mut activation, + &mut report, + ) + }; + assert_eq!( + status, + NemoRelayStatus::Ok, + "activation failed: {:?}", + unsafe { read_last_error() } + ); + assert!(!activation.is_null()); + (activation, unsafe { returned_json(report) }) +} + +fn cstring(value: &str) -> CString { + CString::new(value).expect("C string") +} + +unsafe fn read_last_error() -> Option { + let pointer = nemo_relay_last_error(); + (!pointer.is_null()).then(|| { + unsafe { CStr::from_ptr(pointer) } + .to_string_lossy() + .into_owned() + }) +} + +unsafe fn returned_json(pointer: *mut c_char) -> Json { + assert!(!pointer.is_null(), "expected returned JSON string"); + let json = unsafe { CStr::from_ptr(pointer) } + .to_string_lossy() + .into_owned(); + unsafe { nemo_relay_string_free(pointer) }; + serde_json::from_str(&json).expect("returned JSON") +} + +fn tool_request_intercepts(name: &str, args: Json) -> Json { + let name = cstring(name); + let args = cstring(&args.to_string()); + let mut output = ptr::null_mut(); + let status = unsafe { + api::nemo_relay_tool_request_intercepts(name.as_ptr(), args.as_ptr(), &mut output) + }; + assert_eq!( + status, + NemoRelayStatus::Ok, + "tool request intercept failed: {:?}", + unsafe { read_last_error() } + ); + unsafe { returned_json(output) } +} + +fn plugin_kinds() -> Vec { + let mut output = ptr::null_mut(); + assert_eq!( + unsafe { api::nemo_relay_list_plugin_kinds_json(&mut output) }, + NemoRelayStatus::Ok + ); + serde_json::from_value(unsafe { returned_json(output) }).expect("plugin kinds JSON") +} + +fn build_native_fixture() -> &'static Path { + static FIXTURE: OnceLock = OnceLock::new(); + FIXTURE.get_or_init(|| { + let source_dir = TempDir::new().expect("native fixture source tempdir"); + let fixture_dir = source_dir.path().join("native_plugin"); + let source = fixture_dir.join("src"); + std::fs::create_dir_all(&source).expect("native fixture src dir"); + let plugin_path = Path::new(env!("CARGO_MANIFEST_DIR")).join("../plugin"); + let manifest_template = std::fs::read_to_string( + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../core/tests/fixtures/native_plugin/Cargo.toml"), + ) + .expect("native fixture Cargo.toml"); + let manifest = manifest_template.replace( + r#"nemo-relay-plugin = { path = "../../../../plugin" }"#, + &format!("nemo-relay-plugin = {{ path = {plugin_path:?} }}"), + ); + std::fs::write(fixture_dir.join("Cargo.toml"), manifest) + .expect("write native fixture Cargo.toml"); + std::fs::copy( + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../core/tests/fixtures/native_plugin/src/lib.rs"), + source.join("lib.rs"), + ) + .expect("copy native fixture source"); + + let target = + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/ffi-native-plugin-fixture"); + let status = Command::new(std::env::var("CARGO").unwrap_or_else(|_| "cargo".into())) + .arg("build") + .arg("--quiet") + .arg("--manifest-path") + .arg(fixture_dir.join("Cargo.toml")) + .arg("--target-dir") + .arg(&target) + .status() + .expect("native fixture build should start"); + assert!(status.success(), "native fixture build failed: {status}"); + let library = target.join("debug").join(native_library_name()); + assert!( + library.exists(), + "missing native fixture: {}", + library.display() + ); + library + }) +} + +fn build_worker_fixture() -> &'static Path { + static FIXTURE: OnceLock = OnceLock::new(); + FIXTURE.get_or_init(|| { + let manifest = Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../core/tests/fixtures/worker_plugin/Cargo.toml"); + let target = + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/ffi-worker-plugin-fixture"); + let status = Command::new(std::env::var("CARGO").unwrap_or_else(|_| "cargo".into())) + .arg("build") + .arg("--quiet") + .arg("--locked") + .arg("--manifest-path") + .arg(manifest) + .arg("--target-dir") + .arg(&target) + .status() + .expect("worker fixture build should start"); + assert!(status.success(), "worker fixture build failed: {status}"); + let binary = target.join("debug").join(format!( + "nemo-relay-worker-plugin-fixture{}", + std::env::consts::EXE_SUFFIX + )); + assert!( + binary.exists(), + "missing worker fixture: {}", + binary.display() + ); + binary + }) +} + +fn write_native_manifest(directory: &Path, library: &Path) -> PathBuf { + let manifest = directory.join("relay-plugin.toml"); + std::fs::write( + &manifest, + format!( + r#" +manifest_version = 1 + +[plugin] +id = "fixture_native" +kind = "rust_dynamic" + +[compat] +relay = "={version}" +native_api = "1" + +[defaults] +enabled = false + +[capabilities] +items = ["plugin_native"] + +[load] +library = {library:?} +symbol = "nemo_relay_fixture_native_plugin" +"#, + version = env!("CARGO_PKG_VERSION"), + library = library.to_string_lossy(), + ), + ) + .expect("write native fixture manifest"); + manifest +} + +fn write_worker_manifest(directory: &Path, binary: &Path) -> PathBuf { + let manifest = directory.join("relay-plugin.toml"); + std::fs::write( + &manifest, + format!( + r#" +manifest_version = 1 + +[plugin] +id = "fixture_worker" +kind = "worker" + +[compat] +relay = "={version}" +worker_protocol = "grpc-v1" + +[defaults] +enabled = false + +[capabilities] +items = ["plugin_worker"] + +[load] +runtime = "rust" +entrypoint = {entrypoint:?} +"#, + version = env!("CARGO_PKG_VERSION"), + entrypoint = binary.to_string_lossy(), + ), + ) + .expect("write worker fixture manifest"); + manifest +} + +fn native_library_name() -> &'static str { + if cfg!(target_os = "windows") { + "nemo_relay_plugin_fixture.dll" + } else if cfg!(target_os = "macos") { + "libnemo_relay_plugin_fixture.dylib" + } else { + "libnemo_relay_plugin_fixture.so" + } +} diff --git a/crates/ffi/tests/unit/api/plugin_tests.rs b/crates/ffi/tests/unit/api/plugin_tests.rs index 2d37db847..26c75e904 100644 --- a/crates/ffi/tests/unit/api/plugin_tests.rs +++ b/crates/ffi/tests/unit/api/plugin_tests.rs @@ -5,6 +5,173 @@ use super::*; +#[test] +fn test_ffi_dynamic_plugin_activation_rejects_empty_specs_without_outputs() { + let _guard = TEST_MUTEX.lock().unwrap(); + reset_globals(); + let _ = nemo_relay_clear_plugin_configuration(); + + let config = cstring(r#"{"version":1,"components":[]}"#); + let specs = cstring("[]"); + for _ in 0..2 { + let mut activation = std::ptr::dangling_mut::(); + let mut report_json = std::ptr::dangling_mut::(); + unsafe { + assert_eq!( + nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + &mut activation, + &mut report_json, + ), + NemoRelayStatus::InvalidArg + ); + assert!(activation.is_null()); + assert!(report_json.is_null()); + assert!( + read_last_error() + .unwrap_or_default() + .contains("at least one dynamic plugin") + ); + } + } +} + +#[test] +fn test_ffi_dynamic_plugin_activation_rejects_invalid_inputs_without_outputs() { + let _guard = TEST_MUTEX.lock().unwrap(); + reset_globals(); + let _ = nemo_relay_clear_plugin_configuration(); + + let config = cstring(r#"{"version":1,"components":[]}"#); + let specs = cstring("[]"); + let invalid = cstring("not-json"); + unsafe { + let mut report = std::ptr::dangling_mut::(); + assert_eq!( + nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + ptr::null_mut(), + &mut report, + ), + NemoRelayStatus::NullPointer + ); + assert!(report.is_null()); + + let mut activation = std::ptr::dangling_mut::(); + assert_eq!( + nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + &mut activation, + ptr::null_mut(), + ), + NemoRelayStatus::NullPointer + ); + assert!(activation.is_null()); + + for (config_json, specs_json, expected_error) in [ + (invalid.as_ptr(), specs.as_ptr(), "invalid JSON"), + (config.as_ptr(), invalid.as_ptr(), "invalid JSON"), + ] { + let mut activation = std::ptr::dangling_mut::(); + let mut report = std::ptr::dangling_mut::(); + assert_eq!( + nemo_relay_initialize_with_dynamic_plugins( + config_json, + specs_json, + &mut activation, + &mut report, + ), + NemoRelayStatus::InvalidJson + ); + assert!(activation.is_null()); + assert!(report.is_null()); + assert!( + read_last_error() + .unwrap_or_default() + .contains(expected_error) + ); + } + + let invalid_shape = cstring(r#"{"plugin_id":"not-an-array"}"#); + let mut activation = ptr::null_mut(); + let mut report = ptr::null_mut(); + assert_eq!( + nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + invalid_shape.as_ptr(), + &mut activation, + &mut report, + ), + NemoRelayStatus::InvalidJson + ); + assert!( + read_last_error() + .unwrap_or_default() + .contains("invalid dynamic plugin specifications") + ); + } +} + +#[test] +fn test_ffi_dynamic_plugin_activation_surfaces_load_failures_and_releases_owner() { + let _guard = TEST_MUTEX.lock().unwrap(); + reset_globals(); + let _ = nemo_relay_clear_plugin_configuration(); + + let config = cstring(r#"{"version":1,"components":[]}"#); + let missing_manifest = std::env::temp_dir() + .join(format!("missing-relay-plugin-{}.toml", Uuid::now_v7())) + .to_string_lossy() + .into_owned(); + let specs = cstring( + &json!([{ + "plugin_id": "fixture_missing", + "kind": "rust_dynamic", + "manifest_ref": missing_manifest, + "config": {} + }]) + .to_string(), + ); + + unsafe { + let mut activation = ptr::null_mut(); + let mut report = ptr::null_mut(); + assert_eq!( + nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + &mut activation, + &mut report, + ), + NemoRelayStatus::NotFound + ); + assert!(activation.is_null()); + assert!(report.is_null()); + assert!( + read_last_error() + .unwrap_or_default() + .contains("native plugin load failed") + ); + + let mut retry_activation = ptr::null_mut(); + let mut retry_report = ptr::null_mut(); + assert_eq!( + nemo_relay_initialize_with_dynamic_plugins( + config.as_ptr(), + specs.as_ptr(), + &mut retry_activation, + &mut retry_report, + ), + NemoRelayStatus::NotFound + ); + assert!(retry_activation.is_null()); + assert!(retry_report.is_null()); + } +} + #[test] fn test_ffi_plugin_registration_validation_and_cleanup() { let _guard = TEST_MUTEX.lock().unwrap(); diff --git a/crates/ffi/tests/unit/api_tests.rs b/crates/ffi/tests/unit/api_tests.rs index cf89acd7e..4c8c19c7b 100644 --- a/crates/ffi/tests/unit/api_tests.rs +++ b/crates/ffi/tests/unit/api_tests.rs @@ -17,14 +17,14 @@ use crate::convert::nemo_relay_string_free; use crate::error::{NemoRelayStatus, nemo_relay_last_error}; use crate::types::{ FfiAtifExporter, FfiEvent, FfiLLMHandle, FfiLLMRequest, FfiOpenTelemetrySubscriber, - FfiScopeStack, FfiToolHandle, nemo_relay_atif_exporter_free, nemo_relay_event_data, - nemo_relay_event_input, nemo_relay_event_metadata, nemo_relay_event_model_name, - nemo_relay_event_name, nemo_relay_event_output, nemo_relay_event_parent_uuid, - nemo_relay_event_scope_type, nemo_relay_event_timestamp, nemo_relay_event_tool_call_id, - nemo_relay_event_uuid, nemo_relay_llm_handle_attributes, nemo_relay_llm_handle_free, - nemo_relay_llm_handle_name, nemo_relay_llm_handle_parent_uuid, nemo_relay_llm_handle_uuid, - nemo_relay_llm_request_content, nemo_relay_llm_request_free, nemo_relay_llm_request_headers, - nemo_relay_llm_request_new, nemo_relay_otel_subscriber_free, + FfiPluginActivation, FfiScopeStack, FfiToolHandle, nemo_relay_atif_exporter_free, + nemo_relay_event_data, nemo_relay_event_input, nemo_relay_event_metadata, + nemo_relay_event_model_name, nemo_relay_event_name, nemo_relay_event_output, + nemo_relay_event_parent_uuid, nemo_relay_event_scope_type, nemo_relay_event_timestamp, + nemo_relay_event_tool_call_id, nemo_relay_event_uuid, nemo_relay_llm_handle_attributes, + nemo_relay_llm_handle_free, nemo_relay_llm_handle_name, nemo_relay_llm_handle_parent_uuid, + nemo_relay_llm_handle_uuid, nemo_relay_llm_request_content, nemo_relay_llm_request_free, + nemo_relay_llm_request_headers, nemo_relay_llm_request_new, nemo_relay_otel_subscriber_free, nemo_relay_scope_handle_attributes, nemo_relay_scope_handle_data, nemo_relay_scope_handle_free, nemo_relay_scope_handle_metadata, nemo_relay_scope_handle_name, nemo_relay_scope_handle_parent_uuid, nemo_relay_scope_handle_scope_type, diff --git a/go/nemo_relay/README.md b/go/nemo_relay/README.md index 81baaba83..c7a07459c 100644 --- a/go/nemo_relay/README.md +++ b/go/nemo_relay/README.md @@ -24,6 +24,11 @@ Rust runtime. This binding is experimental and source-first. Rust, Python, and Node.js are the primary supported surfaces. +> **DO NOT TREAT AS PRODUCTION-READY:** the experimental +> `InitializeWithDynamicPlugins` lifecycle needs a real consumer to validate +> shutdown, ownership, and error handling before it can be promoted to a stable +> contract. + ## Why Use It? - **Use NeMo Relay from Go**: Group agent, tool, and LLM work into the same diff --git a/go/nemo_relay/plugin.go b/go/nemo_relay/plugin.go index 1ec43b80e..7da42b475 100644 --- a/go/nemo_relay/plugin.go +++ b/go/nemo_relay/plugin.go @@ -8,6 +8,7 @@ package nemo_relay #include typedef struct FfiPluginContext FfiPluginContext; +typedef struct FfiPluginActivation FfiPluginActivation; typedef void (*NemoRelayFreeFn)(void* user_data); typedef char* (*NemoRelayPluginValidateCb)(void* user_data, const char* plugin_config_json); @@ -27,6 +28,9 @@ typedef char* (*NemoRelayToolExecInterceptCb)(void* user_data, const char* args_ extern int32_t nemo_relay_validate_plugin_config(const char* config_json, char** out_json); extern int32_t nemo_relay_initialize_plugins(const char* config_json, char** out_json); +extern int32_t nemo_relay_initialize_with_dynamic_plugins(const char* config_json, const char* dynamic_plugins_json, FfiPluginActivation** out_activation, char** out_report_json); +extern int32_t nemo_relay_plugin_activation_clear(FfiPluginActivation* activation); +extern void nemo_relay_plugin_activation_free(FfiPluginActivation** activation); extern int32_t nemo_relay_clear_plugin_configuration(void); extern int32_t nemo_relay_active_plugin_report_json(char** out_json); extern int32_t nemo_relay_list_plugin_kinds_json(char** out_json); @@ -68,6 +72,9 @@ import "C" import ( "errors" + "log" + "runtime" + "sync" "unsafe" ) @@ -108,6 +115,43 @@ var ( C.nemo_relay_string_free(out) }) } + initializeWithDynamicPluginsJSON = func(configJSON, dynamicPluginsJSON string) (unsafe.Pointer, string, error) { + cConfig := C.CString(configJSON) + cDynamicPlugins := C.CString(dynamicPluginsJSON) + defer C.free(unsafe.Pointer(cConfig)) + defer C.free(unsafe.Pointer(cDynamicPlugins)) + runtime.LockOSThread() + defer runtime.UnlockOSThread() + + var activation *C.FfiPluginActivation + var report *C.char + status := C.nemo_relay_initialize_with_dynamic_plugins( + cConfig, + cDynamicPlugins, + &activation, + &report, + ) + if err := checkStatus(status); err != nil { + cleanupPartialPluginActivation(activation, report) + return nil, "", err + } + if activation == nil || report == nil { + cleanupPartialPluginActivation(activation, report) + return nil, "", errors.New("dynamic plugin activation returned incomplete outputs") + } + defer C.nemo_relay_string_free(report) + return unsafe.Pointer(activation), C.GoString(report), nil + } + clearPluginActivation = func(ptr unsafe.Pointer) error { + runtime.LockOSThread() + defer runtime.UnlockOSThread() + status := C.nemo_relay_plugin_activation_clear((*C.FfiPluginActivation)(ptr)) + return checkStatus(status) + } + freePluginActivation = func(ptr unsafe.Pointer) { + activation := (*C.FfiPluginActivation)(ptr) + C.nemo_relay_plugin_activation_free(&activation) + } activePluginReportJSON = func() (string, error) { var out *C.char status := C.nemo_relay_active_plugin_report_json(&out) @@ -124,6 +168,16 @@ var ( } ) +func cleanupPartialPluginActivation(activation *C.FfiPluginActivation, report *C.char) { + if report != nil { + C.nemo_relay_string_free(report) + } + if activation != nil { + C.nemo_relay_plugin_activation_clear(activation) + C.nemo_relay_plugin_activation_free(&activation) + } +} + // DiagnosticLevel is the severity level for one plugin diagnostic. type DiagnosticLevel string @@ -176,6 +230,44 @@ type PluginConfig struct { Policy *ConfigPolicy `json:"policy,omitempty"` } +// DynamicPluginKind identifies the runtime lane used by a dynamic plugin. +type DynamicPluginKind string + +const ( + // DynamicPluginKindRustDynamic loads an ABI-compatible native shared library. + DynamicPluginKindRustDynamic DynamicPluginKind = "rust_dynamic" + // DynamicPluginKindWorker starts an isolated worker plugin runtime. + DynamicPluginKindWorker DynamicPluginKind = "worker" +) + +// DynamicPluginActivationSpec describes one explicitly resolved plugin to load. +// ManifestRef and EnvironmentRef are resolved by the embedding application; +// Relay does not perform installation-state discovery through this API. +type DynamicPluginActivationSpec struct { + PluginID string `json:"plugin_id"` + Kind DynamicPluginKind `json:"kind"` + ManifestRef string `json:"manifest_ref"` + EnvironmentRef *string `json:"environment_ref,omitempty"` + Config map[string]any `json:"config,omitempty"` +} + +// PluginActivation owns the runtime registrations, native libraries, and +// workers created by InitializeWithDynamicPlugins. Copies share one activation +// lifetime and may be closed safely from any copy. +// +// Experimental: this API needs a production consumer before its lifecycle +// contract is considered stable. +type PluginActivation struct { + state *pluginActivationState +} + +type pluginActivationState struct { + mu sync.Mutex + ptr unsafe.Pointer + closed bool + closeErr error +} + // PluginContext is the component-scoped registration context passed to plugins. type PluginContext struct { ptr *C.FfiPluginContext @@ -229,6 +321,38 @@ func NewPluginComponent(kind string) PluginComponentSpec { } } +// marshalPluginActivationConfig serializes the activation-only wire shape. +// It keeps presence handling private so the established public Go config +// structs and their encoding method sets remain unchanged. Component enabled +// values are explicit on this wire because Relay defaults an omitted value to +// true, while the Go field's zero value is false. +func marshalPluginActivationConfig(config PluginConfig) ([]byte, error) { + type componentJSON struct { + Kind string `json:"kind"` + Enabled bool `json:"enabled"` + Config map[string]any `json:"config,omitempty"` + } + type configJSON struct { + Version uint32 `json:"version,omitempty"` + Components []componentJSON `json:"components,omitempty"` + Policy *ConfigPolicy `json:"policy,omitempty"` + } + + components := make([]componentJSON, len(config.Components)) + for i, component := range config.Components { + components[i] = componentJSON{ + Kind: component.Kind, + Enabled: component.Enabled, + Config: component.Config, + } + } + return jsonMarshal(configJSON{ + Version: config.Version, + Components: components, + Policy: config.Policy, + }) +} + // ValidatePluginConfig validates a plugin config without changing runtime state. // // It returns the validation report or an error if the config could not be @@ -262,6 +386,101 @@ func InitializePlugins(config PluginConfig) (ConfigReport, error) { return report, nil } +// InitializeWithDynamicPlugins discovers plugins.toml once during startup, layers the +// supplied config over the discovered configuration, and activates explicit +// dynamic plugins in the same transaction. Explicit values take precedence +// where both sources configure the same setting. Static components in the +// resolved configuration are initialized before components appended by dynamic +// plugins. Configuration files are not watched or reloaded. At least one +// dynamic plugin specification is required; use InitializePlugins for a +// static-only configuration. +// +// The caller must retain the returned activation for as long as plugin +// callbacks may run and call Close during orderly shutdown. +// +// Experimental: this API needs a production consumer before its lifecycle +// contract is considered stable. +func InitializeWithDynamicPlugins( + config PluginConfig, + dynamicPlugins []DynamicPluginActivationSpec, +) (*PluginActivation, ConfigReport, error) { + if len(dynamicPlugins) == 0 { + return nil, ConfigReport{}, errors.New( + "dynamic plugin activation requires at least one dynamic plugin; use InitializePlugins for a static-only configuration", + ) + } + configPayload, err := marshalPluginActivationConfig(config) + if err != nil { + return nil, ConfigReport{}, err + } + dynamicPluginsPayload, err := jsonMarshal(dynamicPlugins) + if err != nil { + return nil, ConfigReport{}, err + } + + ptr, rawReport, err := initializeWithDynamicPluginsJSON( + string(configPayload), + string(dynamicPluginsPayload), + ) + if err != nil { + return nil, ConfigReport{}, err + } + activation := newPluginActivation(ptr) + var report ConfigReport + if err := jsonUnmarshal([]byte(rawReport), &report); err != nil { + _ = activation.Close() + return nil, ConfigReport{}, err + } + return activation, report, nil +} + +func newPluginActivation(ptr unsafe.Pointer) *PluginActivation { + state := &pluginActivationState{ptr: ptr} + runtime.SetFinalizer(state, finalizePluginActivation) + return &PluginActivation{state: state} +} + +var reportPluginActivationCleanupError = func(err error) { + log.Printf("nemo_relay: dynamic plugin activation cleanup failed during finalization: %v", err) +} + +func finalizePluginActivation(state *pluginActivationState) { + go func() { + if err := state.close(); err != nil { + reportPluginActivationCleanupError(err) + } + }() +} + +// Close removes callbacks and subscribers before unloading plugin libraries +// and workers. It is safe to call Close repeatedly or on a nil activation. +// Copies and repeated calls observe the same first teardown result. +func (activation *PluginActivation) Close() error { + if activation == nil || activation.state == nil { + return nil + } + return activation.state.close() +} + +func (state *pluginActivationState) close() error { + state.mu.Lock() + defer state.mu.Unlock() + if state.closed { + return state.closeErr + } + state.closed = true + + ptr := state.ptr + state.ptr = nil + if ptr == nil { + return nil + } + runtime.SetFinalizer(state, nil) + state.closeErr = clearPluginActivation(ptr) + freePluginActivation(ptr) + return state.closeErr +} + // ClearPluginConfiguration removes all active plugin component registrations. // // Registered plugin kinds remain available for future validation or diff --git a/go/nemo_relay/plugin_activation_test.go b/go/nemo_relay/plugin_activation_test.go new file mode 100644 index 000000000..c14b88c50 --- /dev/null +++ b/go/nemo_relay/plugin_activation_test.go @@ -0,0 +1,1050 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package nemo_relay + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "regexp" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + "unsafe" +) + +var ( + goNativePluginFixtureOnce sync.Once + goNativePluginFixturePath string + goNativePluginFixtureErr error + goWorkerPluginFixtureOnce sync.Once + goWorkerPluginFixturePath string + goWorkerPluginFixtureErr error + workspacePackagePattern = regexp.MustCompile(`(?ms)^[\t ]*\[workspace\.package\][\t ]*(?:#[^\r\n]*)?\r?\n(.*?)(?:^[\t ]*\[|\z)`) + workspaceVersionPattern = regexp.MustCompile(`(?m)^[\t ]*version[\t ]*=[\t ]*(?:"([^"\r\n]+)"|'([^'\r\n]+)')[\t ]*(?:#[^\r\n]*)?\r?$`) +) + +func withPluginActivationStubs(t *testing.T) { + t.Helper() + originalInitialize := initializeWithDynamicPluginsJSON + originalClear := clearPluginActivation + originalFree := freePluginActivation + originalReporter := reportPluginActivationCleanupError + t.Cleanup(func() { + initializeWithDynamicPluginsJSON = originalInitialize + clearPluginActivation = originalClear + freePluginActivation = originalFree + reportPluginActivationCleanupError = originalReporter + }) +} + +func fixtureDynamicPluginSpecs() []DynamicPluginActivationSpec { + return []DynamicPluginActivationSpec{{ + PluginID: "fixture.native", + Kind: DynamicPluginKindRustDynamic, + ManifestRef: "/tmp/relay-plugin.toml", + }} +} + +func TestInitializeWithDynamicPluginsSerializesSpecsAndOwnsCleanup(t *testing.T) { + withPluginActivationStubs(t) + + token := new(byte) + ptr := unsafe.Pointer(token) + var gotConfig PluginConfig + var gotSpecs []DynamicPluginActivationSpec + initializeWithDynamicPluginsJSON = func(configJSON, specsJSON string) (unsafe.Pointer, string, error) { + if err := json.Unmarshal([]byte(configJSON), &gotConfig); err != nil { + t.Fatalf("invalid config JSON: %v", err) + } + if err := json.Unmarshal([]byte(specsJSON), &gotSpecs); err != nil { + t.Fatalf("invalid specs JSON: %v", err) + } + return ptr, `{"diagnostics":[{"level":"warning","code":"fixture.warning","message":"fixture"}]}`, nil + } + var calls []string + clearPluginActivation = func(got unsafe.Pointer) error { + if got != ptr { + t.Fatalf("clear pointer = %p, want %p", got, ptr) + } + calls = append(calls, "clear") + return nil + } + freePluginActivation = func(got unsafe.Pointer) { + if got != ptr { + t.Fatalf("free pointer = %p, want %p", got, ptr) + } + calls = append(calls, "free") + } + + environment := "/tmp/fixture-environment" + activation, report, err := InitializeWithDynamicPlugins(NewPluginConfig(), []DynamicPluginActivationSpec{ + { + PluginID: "fixture.worker", + Kind: DynamicPluginKindWorker, + ManifestRef: "/tmp/relay-plugin.toml", + EnvironmentRef: &environment, + Config: map[string]any{"tag": "go"}, + }, + }) + if err != nil { + t.Fatalf("InitializeWithDynamicPlugins() error = %v", err) + } + if gotConfig.Version != 1 { + t.Fatalf("config version = %d, want 1", gotConfig.Version) + } + if len(gotSpecs) != 1 || gotSpecs[0].PluginID != "fixture.worker" { + t.Fatalf("specs = %#v", gotSpecs) + } + if gotSpecs[0].EnvironmentRef == nil || *gotSpecs[0].EnvironmentRef != environment { + t.Fatalf("environment ref = %#v", gotSpecs[0].EnvironmentRef) + } + if len(report.Diagnostics) != 1 || report.Diagnostics[0].Code != "fixture.warning" { + t.Fatalf("report = %#v", report) + } + + if err := activation.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + if err := activation.Close(); err != nil { + t.Fatalf("repeated Close() error = %v", err) + } + if strings.Join(calls, ",") != "clear,free" { + t.Fatalf("cleanup calls = %v", calls) + } + runtime.KeepAlive(token) +} + +func TestPluginConfigPublicJSONShapeRemainsCompatible(t *testing.T) { + type envelope struct { + PluginConfig + Name string `json:"name"` + } + + payload, err := json.Marshal(envelope{ + PluginConfig: PluginConfig{Version: 1}, + Name: "fixture", + }) + if err != nil { + t.Fatalf("marshal embedded plugin config: %v", err) + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(payload, &fields); err != nil { + t.Fatalf("unmarshal embedded plugin config: %v", err) + } + if string(fields["name"]) != `"fixture"` || string(fields["version"]) != "1" { + t.Fatalf("embedded plugin config lost envelope fields: %s", payload) + } + + emptyPayload, err := json.Marshal(PluginConfig{}) + if err != nil { + t.Fatalf("marshal empty plugin config: %v", err) + } + if string(emptyPayload) != "{}" { + t.Fatalf("empty plugin config JSON = %s, want {}", emptyPayload) + } +} + +func TestInitializeWithDynamicPluginsUsesPrivateConfigWireShape(t *testing.T) { + withPluginActivationStubs(t) + + config := PluginConfig{Components: []PluginComponentSpec{{Kind: "fixture.disabled"}}} + publicPayload, err := json.Marshal(config) + if err != nil { + t.Fatalf("marshal public plugin config: %v", err) + } + const wantPublic = `{"components":[{"kind":"fixture.disabled"}]}` + if string(publicPayload) != wantPublic { + t.Fatalf("public plugin config JSON = %s, want %s", publicPayload, wantPublic) + } + + token := new(byte) + initializeWithDynamicPluginsJSON = func(configJSON, specsJSON string) (unsafe.Pointer, string, error) { + const wantConfig = `{"components":[{"kind":"fixture.disabled","enabled":false}]}` + if configJSON != wantConfig { + t.Fatalf("config JSON = %s, want %s", configJSON, wantConfig) + } + const wantSpecs = `[{"plugin_id":"fixture.native","kind":"rust_dynamic","manifest_ref":"/tmp/relay-plugin.toml"}]` + if specsJSON != wantSpecs { + t.Fatalf("dynamic plugin specs JSON = %s, want %s", specsJSON, wantSpecs) + } + return unsafe.Pointer(token), `{"diagnostics":[]}`, nil + } + clearPluginActivation = func(unsafe.Pointer) error { return nil } + freePluginActivation = func(unsafe.Pointer) {} + + activation, _, err := InitializeWithDynamicPlugins(config, fixtureDynamicPluginSpecs()) + if err != nil { + t.Fatalf("InitializeWithDynamicPlugins() error = %v", err) + } + if err := activation.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + runtime.KeepAlive(token) +} + +func TestComponentWrappersPreserveDisabledValuesDuringConversion(t *testing.T) { + disabled := []PluginComponentSpec{ + (AdaptiveComponentSpec{Config: NewAdaptiveConfig()}).PluginComponent(), + (ObservabilityComponentSpec{Config: NewObservabilityConfig()}).PluginComponent(), + (PricingComponentSpec{Config: NewPricingConfig()}).PluginComponent(), + (PiiRedactionComponentSpec{Config: NewPiiRedactionConfig()}).PluginComponent(), + } + for _, component := range disabled { + payload, err := marshalPluginActivationConfig(PluginConfig{ + Components: []PluginComponentSpec{component}, + }) + if err != nil { + t.Fatalf("marshal %s component: %v", component.Kind, err) + } + var wire struct { + Components []struct { + Enabled *bool `json:"enabled"` + } `json:"components"` + } + if err := json.Unmarshal(payload, &wire); err != nil { + t.Fatalf("unmarshal %s component wire payload: %v", component.Kind, err) + } + if len(wire.Components) != 1 || wire.Components[0].Enabled == nil || *wire.Components[0].Enabled { + t.Fatalf("%s component wire payload = %s, want enabled false", component.Kind, payload) + } + } +} + +func TestInitializeWithDynamicPluginsRejectsEmptySpecsWithoutCallingCgo(t *testing.T) { + withPluginActivationStubs(t) + + activationCalls := 0 + initializeWithDynamicPluginsJSON = func(string, string) (unsafe.Pointer, string, error) { + activationCalls++ + return nil, "", errors.New("unexpected CGo call") + } + + for _, specs := range [][]DynamicPluginActivationSpec{nil, {}} { + activation, report, err := InitializeWithDynamicPlugins(NewPluginConfig(), specs) + if err == nil || !strings.Contains(err.Error(), "at least one dynamic plugin") { + t.Fatalf("InitializeWithDynamicPlugins(%#v) error = %v, want empty-spec diagnostic", specs, err) + } + if activation != nil || len(report.Diagnostics) != 0 { + t.Fatalf("InitializeWithDynamicPlugins(%#v) = (%#v, %#v), want empty outputs", specs, activation, report) + } + } + if activationCalls != 0 { + t.Fatalf("activation calls = %d, want 0", activationCalls) + } +} + +func TestInitializeWithDynamicPluginsCleansUpInvalidReport(t *testing.T) { + withPluginActivationStubs(t) + + token := new(byte) + ptr := unsafe.Pointer(token) + initializeWithDynamicPluginsJSON = func(string, string) (unsafe.Pointer, string, error) { + return ptr, "not-json", nil + } + var calls []string + clearPluginActivation = func(unsafe.Pointer) error { + calls = append(calls, "clear") + return nil + } + freePluginActivation = func(unsafe.Pointer) { calls = append(calls, "free") } + + activation, _, err := InitializeWithDynamicPlugins(NewPluginConfig(), fixtureDynamicPluginSpecs()) + if err == nil { + t.Fatal("InitializeWithDynamicPlugins() error = nil, want invalid report error") + } + if activation != nil { + t.Fatalf("activation = %#v, want nil", activation) + } + if strings.Join(calls, ",") != "clear,free" { + t.Fatalf("cleanup calls = %v", calls) + } + runtime.KeepAlive(token) +} + +func TestPluginActivationCloseFreesAfterClearFailure(t *testing.T) { + withPluginActivationStubs(t) + + token := new(byte) + ptr := unsafe.Pointer(token) + wantErr := errors.New("teardown failed") + var calls []string + clearPluginActivation = func(unsafe.Pointer) error { + calls = append(calls, "clear") + return wantErr + } + freePluginActivation = func(unsafe.Pointer) { calls = append(calls, "free") } + activation := newPluginActivation(ptr) + + if err := activation.Close(); !errors.Is(err, wantErr) { + t.Fatalf("Close() error = %v, want %v", err, wantErr) + } + if err := activation.Close(); !errors.Is(err, wantErr) { + t.Fatalf("repeated Close() error = %v, want %v", err, wantErr) + } + if strings.Join(calls, ",") != "clear,free" { + t.Fatalf("cleanup calls = %v", calls) + } + runtime.KeepAlive(token) +} + +func TestPluginActivationFinalizerReportsClearFailure(t *testing.T) { + withPluginActivationStubs(t) + + token := new(byte) + wantErr := errors.New("teardown failed") + clearPluginActivation = func(unsafe.Pointer) error { return wantErr } + freePluginActivation = func(unsafe.Pointer) {} + reported := make(chan error, 1) + reportPluginActivationCleanupError = func(err error) { reported <- err } + + finalizePluginActivation(&pluginActivationState{ptr: unsafe.Pointer(token)}) + select { + case err := <-reported: + if !errors.Is(err, wantErr) { + t.Fatalf("reported finalizer error = %v, want %v", err, wantErr) + } + case <-time.After(time.Second): + t.Fatal("finalizer did not report the clear failure") + } + runtime.KeepAlive(token) +} + +func TestPluginActivationFinalizerDoesNotBlockOnCleanup(t *testing.T) { + withPluginActivationStubs(t) + + token := new(byte) + cleanupStarted := make(chan struct{}) + releaseCleanup := make(chan struct{}) + cleanupFinished := make(chan struct{}) + clearPluginActivation = func(unsafe.Pointer) error { + close(cleanupStarted) + <-releaseCleanup + return nil + } + freePluginActivation = func(unsafe.Pointer) { close(cleanupFinished) } + + finalizerReturned := make(chan struct{}) + go func() { + finalizePluginActivation(&pluginActivationState{ptr: unsafe.Pointer(token)}) + close(finalizerReturned) + }() + + select { + case <-cleanupStarted: + case <-time.After(time.Second): + close(releaseCleanup) + t.Fatal("finalizer cleanup did not start") + } + select { + case <-finalizerReturned: + case <-time.After(time.Second): + close(releaseCleanup) + t.Fatal("finalizer blocked while native cleanup was in progress") + } + + close(releaseCleanup) + select { + case <-cleanupFinished: + case <-time.After(time.Second): + t.Fatal("asynchronous finalizer cleanup did not finish") + } + runtime.KeepAlive(token) +} + +func TestPluginActivationCopiesShareCloseStateAndError(t *testing.T) { + withPluginActivationStubs(t) + + token := new(byte) + ptr := unsafe.Pointer(token) + wantErr := errors.New("teardown failed") + var callsMu sync.Mutex + var calls []string + clearPluginActivation = func(got unsafe.Pointer) error { + if got != ptr { + return fmt.Errorf("clear pointer = %p, want %p", got, ptr) + } + callsMu.Lock() + calls = append(calls, "clear") + callsMu.Unlock() + return wantErr + } + freePluginActivation = func(got unsafe.Pointer) { + callsMu.Lock() + defer callsMu.Unlock() + if got != ptr { + calls = append(calls, "free-wrong-pointer") + return + } + calls = append(calls, "free") + } + + activation := newPluginActivation(ptr) + copyValue := *activation + closeErrors := make(chan error, 2) + var closeCalls sync.WaitGroup + for _, handle := range []*PluginActivation{activation, ©Value} { + closeCalls.Add(1) + go func(handle *PluginActivation) { + defer closeCalls.Done() + closeErrors <- handle.Close() + }(handle) + } + closeCalls.Wait() + close(closeErrors) + for err := range closeErrors { + if !errors.Is(err, wantErr) { + t.Fatalf("Close() error = %v, want %v", err, wantErr) + } + } + if err := activation.Close(); !errors.Is(err, wantErr) { + t.Fatalf("repeated Close() error = %v, want %v", err, wantErr) + } + + callsMu.Lock() + gotCalls := strings.Join(calls, ",") + callsMu.Unlock() + if gotCalls != "clear,free" { + t.Fatalf("cleanup calls = %s, want clear,free", gotCalls) + } + runtime.KeepAlive(token) +} + +func TestPluginActivationCopyPreventsEarlyFinalization(t *testing.T) { + withPluginActivationStubs(t) + + token := new(byte) + ptr := unsafe.Pointer(token) + var callsMu sync.Mutex + var calls []string + clearPluginActivation = func(unsafe.Pointer) error { + callsMu.Lock() + calls = append(calls, "clear") + callsMu.Unlock() + return nil + } + freePluginActivation = func(unsafe.Pointer) { + callsMu.Lock() + calls = append(calls, "free") + callsMu.Unlock() + } + + wrapperCollected := make(chan struct{}) + copyValue := copiedPluginActivationWithGCSentinel(ptr, wrapperCollected) + deadline := time.Now().Add(5 * time.Second) + for { + runtime.GC() + runtime.Gosched() + select { + case <-wrapperCollected: + goto wrapperWasCollected + default: + if time.Now().After(deadline) { + t.Fatal("unreachable activation wrapper was not collected") + } + time.Sleep(10 * time.Millisecond) + } + } + +wrapperWasCollected: + for i := 0; i < 3; i++ { + runtime.GC() + runtime.Gosched() + time.Sleep(10 * time.Millisecond) + } + callsMu.Lock() + gotCalls := strings.Join(calls, ",") + callsMu.Unlock() + if gotCalls != "" { + t.Fatalf("cleanup ran while a copied activation was reachable: %s", gotCalls) + } + + if err := copyValue.Close(); err != nil { + t.Fatalf("copied activation Close() error = %v", err) + } + callsMu.Lock() + gotCalls = strings.Join(calls, ",") + callsMu.Unlock() + if gotCalls != "clear,free" { + t.Fatalf("cleanup calls = %s, want clear,free", gotCalls) + } + runtime.KeepAlive(copyValue) + runtime.KeepAlive(token) +} + +type pluginActivationGCSentinel struct { + activation *PluginActivation + padding [64]byte +} + +//go:noinline +func copiedPluginActivationWithGCSentinel( + ptr unsafe.Pointer, + wrapperCollected chan<- struct{}, +) PluginActivation { + activation := newPluginActivation(ptr) + copyValue := *activation + sentinel := &pluginActivationGCSentinel{activation: activation} + runtime.SetFinalizer(sentinel, func(sentinel *pluginActivationGCSentinel) { + runtime.KeepAlive(sentinel.activation) + close(wrapperCollected) + }) + runtime.KeepAlive(activation) + runtime.KeepAlive(sentinel) + return copyValue +} + +func TestInitializeWithDynamicPluginsSurfacesSerializationAndActivationErrors(t *testing.T) { + withPluginActivationStubs(t) + + activationCalls := 0 + initializeWithDynamicPluginsJSON = func(string, string) (unsafe.Pointer, string, error) { + activationCalls++ + return nil, "", errors.New("load failed") + } + + invalidConfig := NewPluginConfig() + invalidConfig.Components = append(invalidConfig.Components, PluginComponentSpec{ + Kind: "fixture", + Enabled: true, + Config: map[string]any{"invalid": make(chan int)}, + }) + if _, _, err := InitializeWithDynamicPlugins(invalidConfig, fixtureDynamicPluginSpecs()); err == nil { + t.Fatal("invalid config serialization error = nil") + } + if activationCalls != 0 { + t.Fatalf("activation calls after config serialization failure = %d", activationCalls) + } + + invalidSpecs := []DynamicPluginActivationSpec{{ + PluginID: "fixture", + Kind: DynamicPluginKindRustDynamic, + ManifestRef: "/tmp/relay-plugin.toml", + Config: map[string]any{"invalid": make(chan int)}, + }} + if _, _, err := InitializeWithDynamicPlugins(NewPluginConfig(), invalidSpecs); err == nil { + t.Fatal("invalid specs serialization error = nil") + } + if activationCalls != 0 { + t.Fatalf("activation calls after specs serialization failure = %d", activationCalls) + } + + if _, _, err := InitializeWithDynamicPlugins(NewPluginConfig(), fixtureDynamicPluginSpecs()); err == nil || err.Error() != "load failed" { + t.Fatalf("activation error = %v, want load failed", err) + } + if activationCalls != 1 { + t.Fatalf("activation calls = %d, want 1", activationCalls) + } +} + +func TestNilPluginActivationCloseIsSafe(t *testing.T) { + var activation *PluginActivation + if err := activation.Close(); err != nil { + t.Fatalf("nil Close() error = %v", err) + } +} + +func TestInitializeWithDynamicPluginsLoadsNativePluginThroughCgo(t *testing.T) { + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration() error = %v", err) + } + library := goNativePluginFixture(t) + manifest := writeGoNativePluginManifest(t, library) + projectDir := t.TempDir() + projectConfigDir := filepath.Join(projectDir, ".nemo-relay") + if err := os.MkdirAll(projectConfigDir, 0o700); err != nil { + t.Fatalf("MkdirAll(project config) error = %v", err) + } + pluginsTOML := filepath.Join(projectConfigDir, "plugins.toml") + const staticKind = "go.fixture.static_base" + fileConfig := fmt.Sprintf(`version = 999 + +[[components]] +kind = %q +enabled = true + +[components.config] +source = "project-file" +`, staticKind) + if err := os.WriteFile(pluginsTOML, []byte(fileConfig), 0o600); err != nil { + t.Fatalf("WriteFile(plugins.toml) error = %v", err) + } + previousCWD, err := os.Getwd() + if err != nil { + t.Fatalf("Getwd() error = %v", err) + } + if err := os.Chdir(projectDir); err != nil { + t.Fatalf("Chdir(project) error = %v", err) + } + t.Cleanup(func() { + if err := os.Chdir(previousCWD); err != nil { + t.Errorf("restore working directory error = %v", err) + } + }) + t.Setenv("XDG_CONFIG_HOME", filepath.Join(projectDir, "xdg")) + + var staticRegistrations atomic.Int32 + var staticCallbacks atomic.Int32 + if err := RegisterPlugin(staticKind, PluginFuncs{ + RegisterFunc: func(config map[string]any, ctx *PluginContext) error { + if config["source"] != "project-file" { + return fmt.Errorf("static plugin config = %#v, want project-file source", config) + } + staticRegistrations.Add(1) + return ctx.RegisterToolRequestIntercept( + "go_static_base", + -1, + false, + func(_ string, args json.RawMessage) json.RawMessage { + staticCallbacks.Add(1) + var payload map[string]any + if err := json.Unmarshal(args, &payload); err != nil { + return args + } + payload["static_saw_dynamic"] = payload["native_plugin"] == true + payload["go_static_base"] = true + out, _ := json.Marshal(payload) + return out + }, + ) + }, + }); err != nil { + t.Fatalf("RegisterPlugin() error = %v", err) + } + defer func() { + if err := DeregisterPlugin(staticKind); err != nil { + t.Errorf("DeregisterPlugin() error = %v", err) + } + }() + + activation, report, err := InitializeWithDynamicPlugins(NewPluginConfig(), []DynamicPluginActivationSpec{{ + PluginID: "fixture_native", + Kind: DynamicPluginKindRustDynamic, + ManifestRef: manifest, + Config: map[string]any{}, + }}) + if err != nil { + t.Fatalf("InitializeWithDynamicPlugins() error = %v", err) + } + defer func() { + if err := activation.Close(); err != nil { + t.Errorf("deferred Close() error = %v", err) + } + }() + if len(report.Diagnostics) != 0 { + t.Fatalf("activation diagnostics = %#v, want none", report.Diagnostics) + } + if got := staticRegistrations.Load(); got != 1 { + t.Fatalf("static registrations = %d, want 1", got) + } + // The activation has already resolved its configuration; subsequent file + // mutations are neither watched nor reloaded. + if err := os.WriteFile(pluginsTOML, []byte("invalid = ["), 0o600); err != nil { + t.Fatalf("mutate plugins.toml error = %v", err) + } + + transformed, err := ToolRequestIntercepts("go-native-tool", json.RawMessage(`{"input":true}`)) + if err != nil { + t.Fatalf("ToolRequestIntercepts() error = %v", err) + } + var transformedObject map[string]any + if err := json.Unmarshal(transformed, &transformedObject); err != nil { + t.Fatalf("transformed tool args are invalid JSON: %v", err) + } + if transformedObject["native_plugin"] != true { + t.Fatalf("transformed tool args = %s, want native_plugin marker", transformed) + } + if transformedObject["go_static_base"] != true { + t.Fatalf("transformed tool args = %s, want static base marker", transformed) + } + if transformedObject["static_saw_dynamic"] != false { + t.Fatalf("transformed tool args = %s, want static callback before dynamic callback", transformed) + } + if got := staticCallbacks.Load(); got != 1 { + t.Fatalf("static callbacks = %d, want 1", got) + } + + if err := activation.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + afterClose, err := ToolRequestIntercepts("go-native-tool", json.RawMessage(`{"input":true}`)) + if err != nil { + t.Fatalf("ToolRequestIntercepts() after Close error = %v", err) + } + if string(afterClose) != `{"input":true}` { + t.Fatalf("tool args after Close = %s, want unchanged args", afterClose) + } + if got := staticCallbacks.Load(); got != 1 { + t.Fatalf("static callbacks after Close = %d, want 1", got) + } + kinds, err := ListPluginKinds() + if err != nil { + t.Fatalf("ListPluginKinds() error = %v", err) + } + for _, kind := range kinds { + if kind == "fixture_native" { + t.Fatal("fixture_native remains registered after Close") + } + } + if err := os.Remove(pluginsTOML); err != nil { + t.Fatalf("Remove(plugins.toml) error = %v", err) + } + + _, _, err = InitializeWithDynamicPlugins(NewPluginConfig(), []DynamicPluginActivationSpec{{ + PluginID: "fixture_missing", + Kind: DynamicPluginKindRustDynamic, + ManifestRef: filepath.Join(t.TempDir(), "missing-relay-plugin.toml"), + }}) + if err == nil || !strings.Contains(err.Error(), "native plugin load failed") { + t.Fatalf("missing-manifest error = %v, want native plugin load diagnostic", err) + } +} + +func TestInitializeWithDynamicPluginsLoadsWorkerPluginThroughCgo(t *testing.T) { + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration() error = %v", err) + } + + executable := goWorkerPluginFixture(t) + manifest := writeGoWorkerPluginManifest(t, executable) + activation, report, err := InitializeWithDynamicPlugins(NewPluginConfig(), []DynamicPluginActivationSpec{{ + PluginID: "fixture_worker", + Kind: DynamicPluginKindWorker, + ManifestRef: manifest, + Config: map[string]any{}, + }}) + if err != nil { + t.Fatalf("InitializeWithDynamicPlugins() error = %v", err) + } + defer func() { + if err := activation.Close(); err != nil { + t.Errorf("deferred Close() error = %v", err) + } + }() + if len(report.Diagnostics) != 0 { + t.Fatalf("activation diagnostics = %#v, want none", report.Diagnostics) + } + + transformed, err := ToolRequestIntercepts("go-worker-tool", json.RawMessage(`{"input":true}`)) + if err != nil { + t.Fatalf("ToolRequestIntercepts() error = %v", err) + } + var transformedObject map[string]any + if err := json.Unmarshal(transformed, &transformedObject); err != nil { + t.Fatalf("transformed tool args are invalid JSON: %v", err) + } + if transformedObject["worker_plugin"] != true { + t.Fatalf("transformed tool args = %s, want worker_plugin marker", transformed) + } + + if err := activation.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + afterClose, err := ToolRequestIntercepts("go-worker-tool", json.RawMessage(`{"input":true}`)) + if err != nil { + t.Fatalf("ToolRequestIntercepts() after Close error = %v", err) + } + if string(afterClose) != `{"input":true}` { + t.Fatalf("tool args after Close = %s, want unchanged args", afterClose) + } +} + +func TestPluginActivationFinalizerReleasesHostOwnership(t *testing.T) { + if err := ClearPluginConfiguration(); err != nil { + t.Fatalf("ClearPluginConfiguration() error = %v", err) + } + library := goNativePluginFixture(t) + manifest := writeGoNativePluginManifest(t, library) + specs := []DynamicPluginActivationSpec{{ + PluginID: "fixture_native", + Kind: DynamicPluginKindRustDynamic, + ManifestRef: manifest, + Config: map[string]any{}, + }} + createUnclosedPluginActivation(t, specs) + + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + runtime.GC() + runtime.Gosched() + activation, _, err := InitializeWithDynamicPlugins(NewPluginConfig(), specs) + if err == nil { + if closeErr := activation.Close(); closeErr != nil { + t.Fatalf("replacement activation Close() error = %v", closeErr) + } + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("plugin activation finalizer did not release host ownership") +} + +// Keep creation in a separate frame so the activation is unreachable when the +// caller starts forcing collection. +// +//go:noinline +func createUnclosedPluginActivation(t *testing.T, specs []DynamicPluginActivationSpec) { + t.Helper() + activation, _, err := InitializeWithDynamicPlugins(NewPluginConfig(), specs) + if err != nil { + t.Fatalf("InitializeWithDynamicPlugins() error = %v", err) + } + runtime.KeepAlive(activation) +} + +func goNativePluginFixture(t *testing.T) string { + t.Helper() + goNativePluginFixtureOnce.Do(func() { + repoRoot, err := filepath.Abs(filepath.Join("..", "..")) + if err != nil { + goNativePluginFixtureErr = err + return + } + sourceRoot, err := os.MkdirTemp("", "nemo-relay-go-native-plugin-") + if err != nil { + goNativePluginFixtureErr = err + return + } + defer os.RemoveAll(sourceRoot) + fixtureRoot := filepath.Join(sourceRoot, "native_plugin") + if err := os.MkdirAll(filepath.Join(fixtureRoot, "src"), 0o700); err != nil { + goNativePluginFixtureErr = err + return + } + fixtureSource := filepath.Join(repoRoot, "crates", "core", "tests", "fixtures", "native_plugin") + manifestBytes, err := os.ReadFile(filepath.Join(fixtureSource, "Cargo.toml")) + if err != nil { + goNativePluginFixtureErr = err + return + } + pluginPath := filepath.Join(repoRoot, "crates", "plugin") + manifestContents := strings.Replace( + string(manifestBytes), + `nemo-relay-plugin = { path = "../../../../plugin" }`, + fmt.Sprintf("nemo-relay-plugin = { path = %q }", pluginPath), + 1, + ) + manifest := filepath.Join(fixtureRoot, "Cargo.toml") + if err := os.WriteFile(manifest, []byte(manifestContents), 0o600); err != nil { + goNativePluginFixtureErr = err + return + } + librarySource, err := os.ReadFile(filepath.Join(fixtureSource, "src", "lib.rs")) + if err != nil { + goNativePluginFixtureErr = err + return + } + if err := os.WriteFile(filepath.Join(fixtureRoot, "src", "lib.rs"), librarySource, 0o600); err != nil { + goNativePluginFixtureErr = err + return + } + target := filepath.Join(repoRoot, "target") + cargo := os.Getenv("CARGO") + if cargo == "" { + cargo = "cargo" + } + command := exec.Command(cargo, "build", "--quiet", "--manifest-path", manifest, "--target-dir", target) + if output, err := command.CombinedOutput(); err != nil { + goNativePluginFixtureErr = fmt.Errorf("build native plugin fixture: %w\n%s", err, output) + return + } + goNativePluginFixturePath = filepath.Join(target, "debug", goNativeLibraryName()) + if _, err := os.Stat(goNativePluginFixturePath); err != nil { + goNativePluginFixtureErr = fmt.Errorf("native plugin fixture output: %w", err) + } + }) + if goNativePluginFixtureErr != nil { + t.Fatal(goNativePluginFixtureErr) + } + return goNativePluginFixturePath +} + +func goWorkerPluginFixture(t *testing.T) string { + t.Helper() + goWorkerPluginFixtureOnce.Do(func() { + repoRoot, err := filepath.Abs(filepath.Join("..", "..")) + if err != nil { + goWorkerPluginFixtureErr = err + return + } + manifest := filepath.Join(repoRoot, "crates", "core", "tests", "fixtures", "worker_plugin", "Cargo.toml") + target := filepath.Join(repoRoot, "target") + cargo := os.Getenv("CARGO") + if cargo == "" { + cargo = "cargo" + } + command := exec.Command(cargo, "build", "--quiet", "--locked", "--manifest-path", manifest, "--target-dir", target) + if output, err := command.CombinedOutput(); err != nil { + goWorkerPluginFixtureErr = fmt.Errorf("build worker plugin fixture: %w\n%s", err, output) + return + } + executable := "nemo-relay-worker-plugin-fixture" + if runtime.GOOS == "windows" { + executable += ".exe" + } + goWorkerPluginFixturePath = filepath.Join(target, "debug", executable) + if _, err := os.Stat(goWorkerPluginFixturePath); err != nil { + goWorkerPluginFixtureErr = fmt.Errorf("worker plugin fixture output: %w", err) + } + }) + if goWorkerPluginFixtureErr != nil { + t.Fatal(goWorkerPluginFixtureErr) + } + return goWorkerPluginFixturePath +} + +func writeGoNativePluginManifest(t *testing.T, library string) string { + t.Helper() + manifest := filepath.Join(t.TempDir(), "relay-plugin.toml") + contents := fmt.Sprintf(`manifest_version = 1 + +[plugin] +id = "fixture_native" +kind = "rust_dynamic" + +[compat] +relay = "=%s" +native_api = "1" + +[defaults] +enabled = false + +[capabilities] +items = ["plugin_native"] + +[load] +library = %q +symbol = "nemo_relay_fixture_native_plugin" +`, relayWorkspaceVersion(t), library) + if err := os.WriteFile(manifest, []byte(contents), 0o600); err != nil { + t.Fatalf("write native plugin manifest: %v", err) + } + return manifest +} + +func writeGoWorkerPluginManifest(t *testing.T, executable string) string { + t.Helper() + manifest := filepath.Join(t.TempDir(), "relay-plugin.toml") + contents := fmt.Sprintf(`manifest_version = 1 + +[plugin] +id = "fixture_worker" +kind = "worker" + +[compat] +relay = "=%s" +worker_protocol = "grpc-v1" + +[defaults] +enabled = false + +[capabilities] +items = ["plugin_worker"] + +[load] +runtime = "rust" +entrypoint = %q +`, relayWorkspaceVersion(t), executable) + if err := os.WriteFile(manifest, []byte(contents), 0o600); err != nil { + t.Fatalf("write worker plugin manifest: %v", err) + } + return manifest +} + +func relayWorkspaceVersion(t *testing.T) string { + t.Helper() + repoRoot, err := filepath.Abs(filepath.Join("..", "..")) + if err != nil { + t.Fatalf("resolve repository root: %v", err) + } + payload, err := os.ReadFile(filepath.Join(repoRoot, "Cargo.toml")) + if err != nil { + t.Fatalf("read workspace Cargo.toml: %v", err) + } + version, err := workspaceVersionFromCargoTOML(payload) + if err != nil { + t.Fatal(err) + } + return version +} + +func workspaceVersionFromCargoTOML(payload []byte) (string, error) { + section := workspacePackagePattern.FindSubmatch(payload) + if section == nil { + return "", errors.New("workspace Cargo.toml has no [workspace.package] section") + } + version := workspaceVersionPattern.FindSubmatch(section[1]) + if version == nil { + return "", errors.New("workspace package version not found") + } + if len(version[1]) != 0 { + return string(version[1]), nil + } + return string(version[2]), nil +} + +func TestWorkspaceVersionFromCargoTOML(t *testing.T) { + tests := []struct { + name string + payload string + want string + wantErr string + }{ + { + name: "standard workspace package", + payload: "[workspace.package]\nversion = \"0.6.0\"\nedition = \"2024\"\n\n[workspace.dependencies]\n", + want: "0.6.0", + }, + { + name: "whitespace comments CRLF and literal string", + payload: "[package]\r\nversion = \"9.9.9\"\r\n\r\n [workspace.package] # inherited metadata\r\n version\t=\t'0.7.0' # next release\r\n\r\n[[workspace.metadata.fixture]]\r\n", + want: "0.7.0", + }, + { + name: "missing workspace package", + payload: "[package]\nversion = \"0.6.0\"\n", + wantErr: "no [workspace.package] section", + }, + { + name: "missing workspace version", + payload: "[workspace.package]\nedition = \"2024\"\n", + wantErr: "workspace package version not found", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := workspaceVersionFromCargoTOML([]byte(test.payload)) + if test.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), test.wantErr) { + t.Fatalf("workspaceVersionFromCargoTOML() error = %v, want %q", err, test.wantErr) + } + return + } + if err != nil { + t.Fatalf("workspaceVersionFromCargoTOML() error = %v", err) + } + if got != test.want { + t.Fatalf("workspaceVersionFromCargoTOML() = %q, want %q", got, test.want) + } + }) + } +} + +func goNativeLibraryName() string { + switch runtime.GOOS { + case "windows": + return "nemo_relay_plugin_fixture.dll" + case "darwin": + return "libnemo_relay_plugin_fixture.dylib" + default: + return "libnemo_relay_plugin_fixture.so" + } +}