From 0661bcd07b99a50adac2c3a47b3d62cbc785cb95 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Sat, 18 Jul 2026 13:17:25 -0400 Subject: [PATCH 1/2] fix: make Node middleware callback errors catchable Signed-off-by: Will Killian --- crates/node/src/api/mod.rs | 245 +++++++++++++----- crates/node/src/callable.rs | 191 +++++++++----- crates/node/src/types/mod.rs | 49 +--- crates/node/tests/adaptive_tests.mjs | 34 +++ crates/node/tests/callback_error_tests.mjs | 11 +- crates/node/tests/event_sanitizers_tests.mjs | 49 ++++ crates/node/tests/llm_tests.mjs | 55 ++++ crates/node/tests/scope_local_tests.mjs | 32 +++ crates/node/tests/tools_tests.mjs | 93 +++++++ .../advanced-guide.mdx | 8 + docs/reference/event-sanitizers.mdx | 5 +- 11 files changed, 607 insertions(+), 165 deletions(-) diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 1a7cc8a5e..94925f466 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -78,7 +78,7 @@ use crate::convert::{ use crate::stream::LlmStream; use crate::types::{ EventSanitizeFields, LlmHandle, ScopeHandle, ScopeStack, ScopeType, ToolHandle, - event_sanitize_fields_from_js, + event_sanitize_fields_from_json, }; #[napi::module_init] @@ -564,6 +564,41 @@ fn json_callback_tsfn( Ok(tsfn) } +fn middleware_tool_callback_tsfn( + env: &Env, + func: &JsFunction, +) -> napi::Result> { + let callback = callable::safe_middleware_callback(env, func)?; + let mut tsfn = callback.create_threadsafe_function( + 0, + |ctx: napi::threadsafe_function::ThreadSafeCallContext<(String, Json)>| { + let name = ctx.env.create_string_from_std(ctx.value.0)?; + let args = unsafe { + JsUnknown::from_raw_unchecked( + ctx.env.raw(), + Json::to_napi_value(ctx.env.raw(), ctx.value.1)?, + ) + }; + Ok(vec![js_unknown_from_raw(&ctx.env, &name), args]) + }, + )?; + tsfn.unref(env)?; + Ok(tsfn) +} + +fn middleware_json_callback_tsfn( + env: &Env, + func: &JsFunction, +) -> napi::Result> { + let callback = callable::safe_middleware_callback(env, func)?; + let mut tsfn = callback.create_threadsafe_function( + 0, + |ctx: napi::threadsafe_function::ThreadSafeCallContext| Ok(vec![ctx.value]), + )?; + tsfn.unref(env)?; + Ok(tsfn) +} + #[allow(clippy::too_many_arguments)] fn add_plugin_event_sanitizer( env: &Env, @@ -683,12 +718,11 @@ fn build_plugin_context( ctx.get::(0)? ); let priority = ctx.get::(1)?; - let callback = - ctx.get::>(2)?; + let callback = ctx.get::(2)?; core_registry_api::register_tool_sanitize_request_guardrail( &name, priority, - callable::wrap_js_tool_fn(callback), + callable::wrap_js_tool_fn(middleware_tool_callback_tsfn(ctx.env, &callback)?), ) .map_err(to_napi_err)?; @@ -728,12 +762,11 @@ fn build_plugin_context( ctx.get::(0)? ); let priority = ctx.get::(1)?; - let callback = - ctx.get::>(2)?; + let callback = ctx.get::(2)?; core_registry_api::register_tool_sanitize_response_guardrail( &name, priority, - callable::wrap_js_tool_fn(callback), + callable::wrap_js_tool_fn(middleware_tool_callback_tsfn(ctx.env, &callback)?), ) .map_err(to_napi_err)?; @@ -769,12 +802,13 @@ fn build_plugin_context( move |ctx| { let name = format!("{}{}", tool_conditional_namespace, ctx.get::(0)?); let priority = ctx.get::(1)?; - let callback = - ctx.get::>(2)?; + let callback = ctx.get::(2)?; core_registry_api::register_tool_conditional_execution_guardrail( &name, priority, - callable::wrap_js_tool_conditional_fn(callback), + callable::wrap_js_tool_conditional_fn(middleware_tool_callback_tsfn( + ctx.env, &callback, + )?), ) .map_err(to_napi_err)?; @@ -816,11 +850,13 @@ fn build_plugin_context( ctx.get::(0)? ); let priority = ctx.get::(1)?; - let callback = ctx.get::>(2)?; + let callback = ctx.get::(2)?; core_registry_api::register_llm_sanitize_request_guardrail( &name, priority, - callable::wrap_js_llm_sanitize_request_fn(callback), + callable::wrap_js_llm_sanitize_request_fn(middleware_json_callback_tsfn( + ctx.env, &callback, + )?), ) .map_err(to_napi_err)?; @@ -860,11 +896,13 @@ fn build_plugin_context( ctx.get::(0)? ); let priority = ctx.get::(1)?; - let callback = ctx.get::>(2)?; + let callback = ctx.get::(2)?; core_registry_api::register_llm_sanitize_response_guardrail( &name, priority, - callable::wrap_js_llm_response_fn(callback), + callable::wrap_js_llm_response_fn(middleware_json_callback_tsfn( + ctx.env, &callback, + )?), ) .map_err(to_napi_err)?; @@ -900,11 +938,13 @@ fn build_plugin_context( move |ctx| { let name = format!("{}{}", llm_conditional_namespace, ctx.get::(0)?); let priority = ctx.get::(1)?; - let callback = ctx.get::>(2)?; + let callback = ctx.get::(2)?; core_registry_api::register_llm_conditional_execution_guardrail( &name, priority, - callable::wrap_js_llm_conditional_fn(callback), + callable::wrap_js_llm_conditional_fn(middleware_json_callback_tsfn( + ctx.env, &callback, + )?), ) .map_err(to_napi_err)?; @@ -944,7 +984,7 @@ fn build_plugin_context( let priority = ctx.get::(1)?; let break_chain = ctx.get::(2)?; let callback = ctx.get::(3)?; - let tsfn = json_callback_tsfn(ctx.env, &callback)?; + let tsfn = middleware_json_callback_tsfn(ctx.env, &callback)?; core_registry_api::register_llm_request_intercept( &name, priority, @@ -1072,8 +1112,8 @@ fn build_plugin_context( let name = format!("{}{}", tool_request_namespace, ctx.get::(0)?); let priority = ctx.get::(1)?; let break_chain = ctx.get::(2)?; - let callback = - ctx.get::>(3)?; + let callback = ctx.get::(3)?; + let callback = middleware_tool_callback_tsfn(ctx.env, &callback)?; core_registry_api::register_tool_request_intercept( &name, priority, @@ -1230,11 +1270,7 @@ impl PersistentJsFunction { unsafe { Option::::from_napi_value(self.env, returned.raw()) }.map(callback_json) } - fn call_event_sanitize( - &self, - event: Json, - fields: EventSanitizeFields, - ) -> napi::Result { + fn call_event_sanitize(&self, event: Json, fields: EventSanitizeFields) -> napi::Result { let mut value = ptr::null_mut(); // SAFETY: `self.reference` is a live N-API reference created in // `self.env`, and `value` is writable storage for the borrowed @@ -1264,7 +1300,8 @@ impl PersistentJsFunction { ) }; let returned = func.call(None, &[event, fields])?; - event_sanitize_fields_from_js(returned) + // SAFETY: `returned` is the live result of invoking `func` in this environment. + unsafe { Option::::from_napi_value(self.env, returned.raw()) }.map(callback_json) } } @@ -1294,9 +1331,10 @@ fn js_event_fields(fields: &nemo_relay::api::event::EventSanitizeFields) -> Even } fn node_event_sanitize_fn(env: &Env, func: &JsFunction) -> napi::Result { - let direct = Arc::new(PersistentJsFunction::new(env, func)?); + let callback = callable::safe_middleware_callback(env, func)?; + let direct = Arc::new(PersistentJsFunction::new(env, &callback)?); let register_thread = std::thread::current().id(); - let mut tsfn = func.create_threadsafe_function( + let mut tsfn = callback.create_threadsafe_function( 0, |ctx: napi::threadsafe_function::ThreadSafeCallContext<(Json, Json)>| { Ok(vec![ctx.value.0, ctx.value.1]) @@ -1315,14 +1353,37 @@ fn node_event_sanitize_fn(env: &Env, func: &JsFunction) -> napi::Result FlowResult<_> { + let value = direct + .call_event_sanitize(event_json, js_event_fields(&fields)) + .map_err(|error| { + FlowError::Internal(format!( + "nemo_relay: JS event sanitizer callback failed: {error}" + )) + })?; + let value = callable::unwrap_middleware_result( + value, + "nemo_relay: JS event sanitizer callback failed", + )?; + let fields = event_sanitize_fields_from_json(value).map_err(|error| { + FlowError::Internal(format!( + "nemo_relay: JS event sanitizer callback failed: invalid JS event sanitizer result: {error}" + )) + })?; + core_event_fields(fields).ok_or_else(|| { + FlowError::Internal( + "nemo_relay: JS event sanitizer callback failed: invalid JS event sanitizer result" + .to_string(), + ) + }) + })(); + match sanitized { + Ok(sanitized) => sanitized, + Err(error) => { + record_callback_error(error.to_string()); + fields + } } - sanitized.unwrap_or(fields) } else { background(event, fields) } @@ -1462,8 +1523,8 @@ pub fn scope_stack_active() -> bool { /// Returns the most recent callback error that could not be surfaced through a direct exception. /// -/// This is primarily used for sanitize/intercept/finalizer callback paths whose -/// core callback signatures cannot return `Result`. +/// This is primarily used for sanitize callback paths that fail open and cannot +/// surface their errors directly. #[napi] pub fn get_last_callback_error() -> Option { get_recorded_callback_error() @@ -2319,6 +2380,10 @@ pub fn llm_stream_call_execute( macro_rules! napi_event_guardrail_api { ($register_name:ident, $deregister_name:ident, $core_register:path, $core_deregister:path) => { + /// Register an event sanitize guardrail. + /// + /// The callback must be synchronous. If it throws, Relay preserves the current event + /// fields and records the error for `getLastCallbackError()`. #[napi] pub fn $register_name( env: Env, @@ -2366,11 +2431,13 @@ macro_rules! napi_guardrail_tool_api { $(#[doc = $reg_doc])* #[napi] pub fn $register_name( + env: Env, name: String, priority: i32, - guardrail: ThreadsafeFunction<(String, Json), ErrorStrategy::Fatal>, + guardrail: JsFunction, ) -> Result<()> { - $core_register(&name, priority, $wrapper(guardrail)).map_err(to_napi_err) + let callback = middleware_tool_callback_tsfn(&env, &guardrail)?; + $core_register(&name, priority, $wrapper(callback)).map_err(to_napi_err) } $(#[doc = $dereg_doc])* @@ -2386,6 +2453,8 @@ napi_guardrail_tool_api!( /// /// The `guardrail` callback receives `(toolName, args)` and must return sanitized args. /// Higher `priority` values run first. Throws if a guardrail with the same `name` already exists. + /// If the callback throws, Relay preserves the current emitted payload and records the error + /// for `getLastCallbackError()`. register_tool_sanitize_request_guardrail, /// Deregister a tool request sanitization guardrail by name. /// @@ -2401,6 +2470,8 @@ napi_guardrail_tool_api!( /// /// The `guardrail` callback receives `(toolName, result)` and must return sanitized result. /// Higher `priority` values run first. Throws if a guardrail with the same `name` already exists. + /// If the callback throws, Relay preserves the current emitted payload and records the error + /// for `getLastCallbackError()`. register_tool_sanitize_response_guardrail, /// Deregister a tool response sanitization guardrail by name. /// @@ -2415,16 +2486,19 @@ napi_guardrail_tool_api!( /// /// The `guardrail` callback receives `(toolName, args)` and must return `null` to allow /// execution or a rejection reason string to block it. Higher `priority` values run first. +/// If the callback throws, the managed call rejects and the protected callback does not run. #[napi] pub fn register_tool_conditional_execution_guardrail( + env: Env, name: String, priority: i32, - guardrail: ThreadsafeFunction<(String, Json), ErrorStrategy::Fatal>, + guardrail: JsFunction, ) -> Result<()> { + let callback = middleware_tool_callback_tsfn(&env, &guardrail)?; core_registry_api::register_tool_conditional_execution_guardrail( &name, priority, - callable::wrap_js_tool_conditional_fn(guardrail), + callable::wrap_js_tool_conditional_fn(callback), ) .map_err(to_napi_err) } @@ -2448,12 +2522,14 @@ macro_rules! napi_intercept_tool_api { $(#[doc = $reg_doc])* #[napi] pub fn $register_name( + env: Env, name: String, priority: i32, break_chain: bool, - callable: ThreadsafeFunction<(String, Json), ErrorStrategy::Fatal>, + callable: JsFunction, ) -> Result<()> { - $core_register(&name, priority, break_chain, $wrapper(callable)).map_err(to_napi_err) + let callback = middleware_tool_callback_tsfn(&env, &callable)?; + $core_register(&name, priority, break_chain, $wrapper(callback)).map_err(to_napi_err) } $(#[doc = $dereg_doc])* @@ -2469,6 +2545,7 @@ napi_intercept_tool_api!( /// /// The `callable` receives `(toolName, args)` and returns transformed args. If `breakChain` /// is `true`, no lower-priority intercepts run after this one. Higher `priority` values run first. + /// If the callback throws, the managed call rejects and later middleware does not run. register_tool_request_intercept, /// Deregister a tool request intercept by name. /// @@ -2532,16 +2609,20 @@ pub fn deregister_tool_execution_intercept(name: String) -> Result { /// /// The `guardrail` callback receives the LLM request as JSON and must return the sanitized request. /// Higher `priority` values run first. Throws if a guardrail with the same `name` already exists. +/// If the callback throws, Relay preserves the current emitted payload and records the error for +/// `getLastCallbackError()`. #[napi] pub fn register_llm_sanitize_request_guardrail( + env: Env, name: String, priority: i32, - guardrail: ThreadsafeFunction, + guardrail: JsFunction, ) -> Result<()> { + let callback = middleware_json_callback_tsfn(&env, &guardrail)?; core_registry_api::register_llm_sanitize_request_guardrail( &name, priority, - callable::wrap_js_llm_sanitize_request_fn(guardrail), + callable::wrap_js_llm_sanitize_request_fn(callback), ) .map_err(to_napi_err) } @@ -2559,16 +2640,20 @@ pub fn deregister_llm_sanitize_request_guardrail(name: String) -> Result { /// The `guardrail` callback receives the LLM response as a JSON value and must return /// the sanitized response as JSON. Higher `priority` values run first. Throws if a guardrail /// with the same `name` already exists. +/// If the callback throws, Relay preserves the current emitted payload and records the error for +/// `getLastCallbackError()`. #[napi] pub fn register_llm_sanitize_response_guardrail( + env: Env, name: String, priority: i32, - guardrail: ThreadsafeFunction, + guardrail: JsFunction, ) -> Result<()> { + let callback = middleware_json_callback_tsfn(&env, &guardrail)?; core_registry_api::register_llm_sanitize_response_guardrail( &name, priority, - callable::wrap_js_llm_response_fn(guardrail), + callable::wrap_js_llm_response_fn(callback), ) .map_err(to_napi_err) } @@ -2585,16 +2670,19 @@ pub fn deregister_llm_sanitize_response_guardrail(name: String) -> Result /// /// The `guardrail` callback receives the LLM request as JSON and must return `null` to allow /// execution or a rejection reason string to block it. Higher `priority` values run first. +/// If the callback throws, the managed call rejects and the protected callback does not run. #[napi] pub fn register_llm_conditional_execution_guardrail( + env: Env, name: String, priority: i32, - guardrail: ThreadsafeFunction, + guardrail: JsFunction, ) -> Result<()> { + let callback = middleware_json_callback_tsfn(&env, &guardrail)?; core_registry_api::register_llm_conditional_execution_guardrail( &name, priority, - callable::wrap_js_llm_conditional_fn(guardrail), + callable::wrap_js_llm_conditional_fn(callback), ) .map_err(to_napi_err) } @@ -2616,21 +2704,24 @@ pub fn deregister_llm_conditional_execution_guardrail(name: String) -> Result, + callable: JsFunction, ) -> Result<()> { + let callback = middleware_json_callback_tsfn(&env, &callable)?; core_registry_api::register_llm_request_intercept( &name, priority, break_chain, - callable::wrap_js_llm_request_intercept_fn(callable), + callable::wrap_js_llm_request_intercept_fn(callback), ) .map_err(to_napi_err) } @@ -2770,6 +2861,10 @@ pub fn flush_subscribers() -> Result<()> { macro_rules! napi_scope_event_guardrail_api { ($register_name:ident, $deregister_name:ident, $core_register:path, $core_deregister:path) => { + /// Register a scope-local event sanitize guardrail. + /// + /// The callback must be synchronous. If it throws, Relay preserves the current event + /// fields and records the error for `getLastCallbackError()`. #[napi] pub fn $register_name( env: Env, @@ -2827,14 +2922,16 @@ macro_rules! napi_scope_guardrail_tool_api { $(#[doc = $reg_doc])* #[napi] pub fn $register_name( + env: Env, scope_uuid: String, name: String, priority: i32, - guardrail: ThreadsafeFunction<(String, Json), ErrorStrategy::Fatal>, + guardrail: JsFunction, ) -> Result<()> { let uuid = uuid::Uuid::parse_str(&scope_uuid) .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; - $core_register(&uuid, &name, priority, $wrapper(guardrail)).map_err(to_napi_err) + let callback = middleware_tool_callback_tsfn(&env, &guardrail)?; + $core_register(&uuid, &name, priority, $wrapper(callback)).map_err(to_napi_err) } $(#[doc = $dereg_doc])* @@ -2853,6 +2950,8 @@ napi_scope_guardrail_tool_api!( /// The `guardrail` callback receives `(toolName, args)` and must return sanitized args. /// Higher `priority` values run first. Throws if a guardrail with the same `name` already exists /// on the specified scope. + /// If the callback throws, Relay preserves the current emitted payload and records the error + /// for `getLastCallbackError()`. scope_register_tool_sanitize_request_guardrail, /// Deregister a scope-local tool request sanitization guardrail by name. /// @@ -2869,6 +2968,8 @@ napi_scope_guardrail_tool_api!( /// The `guardrail` callback receives `(toolName, result)` and must return sanitized result. /// Higher `priority` values run first. Throws if a guardrail with the same `name` already exists /// on the specified scope. + /// If the callback throws, Relay preserves the current emitted payload and records the error + /// for `getLastCallbackError()`. scope_register_tool_sanitize_response_guardrail, /// Deregister a scope-local tool response sanitization guardrail by name. /// @@ -2883,12 +2984,14 @@ napi_scope_guardrail_tool_api!( /// /// The `guardrail` callback receives `(toolName, args)` and must return `null` to allow /// execution or a rejection reason string to block it. Higher `priority` values run first. +/// If the callback throws, the managed call rejects and the protected callback does not run. #[napi] pub fn scope_register_tool_conditional_execution_guardrail( + env: Env, scope_uuid: String, name: String, priority: i32, - guardrail: ThreadsafeFunction<(String, Json), ErrorStrategy::Fatal>, + guardrail: JsFunction, ) -> Result<()> { let uuid = uuid::Uuid::parse_str(&scope_uuid) .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; @@ -2896,7 +2999,7 @@ pub fn scope_register_tool_conditional_execution_guardrail( &uuid, &name, priority, - callable::wrap_js_tool_conditional_fn(guardrail), + callable::wrap_js_tool_conditional_fn(middleware_tool_callback_tsfn(&env, &guardrail)?), ) .map_err(to_napi_err) } @@ -2926,15 +3029,17 @@ macro_rules! napi_scope_intercept_tool_api { $(#[doc = $reg_doc])* #[napi] pub fn $register_name( + env: Env, scope_uuid: String, name: String, priority: i32, break_chain: bool, - callable: ThreadsafeFunction<(String, Json), ErrorStrategy::Fatal>, + callable: JsFunction, ) -> Result<()> { let uuid = uuid::Uuid::parse_str(&scope_uuid) .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; - $core_register(&uuid, &name, priority, break_chain, $wrapper(callable)) + let callback = middleware_tool_callback_tsfn(&env, &callable)?; + $core_register(&uuid, &name, priority, break_chain, $wrapper(callback)) .map_err(to_napi_err) } @@ -2953,6 +3058,7 @@ napi_scope_intercept_tool_api!( /// /// The `callable` receives `(toolName, args)` and returns transformed args. If `breakChain` /// is `true`, no lower-priority intercepts run after this one. Higher `priority` values run first. + /// If the callback throws, the managed call rejects and later middleware does not run. scope_register_tool_request_intercept, /// Deregister a scope-local tool request intercept by name. /// @@ -3029,12 +3135,15 @@ pub fn scope_deregister_tool_execution_intercept(scope_uuid: String, name: Strin /// The `guardrail` callback receives the LLM request as JSON and must return the sanitized request. /// Higher `priority` values run first. Throws if a guardrail with the same `name` already exists /// on the specified scope. +/// If the callback throws, Relay preserves the current emitted payload and records the error for +/// `getLastCallbackError()`. #[napi] pub fn scope_register_llm_sanitize_request_guardrail( + env: Env, scope_uuid: String, name: String, priority: i32, - guardrail: ThreadsafeFunction, + guardrail: JsFunction, ) -> Result<()> { let uuid = uuid::Uuid::parse_str(&scope_uuid) .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; @@ -3042,7 +3151,7 @@ pub fn scope_register_llm_sanitize_request_guardrail( &uuid, &name, priority, - callable::wrap_js_llm_sanitize_request_fn(guardrail), + callable::wrap_js_llm_sanitize_request_fn(middleware_json_callback_tsfn(&env, &guardrail)?), ) .map_err(to_napi_err) } @@ -3066,12 +3175,15 @@ pub fn scope_deregister_llm_sanitize_request_guardrail( /// The `guardrail` callback receives the LLM response as a JSON value and must return /// the sanitized response as JSON. Higher `priority` values run first. Throws if a guardrail /// with the same `name` already exists on the specified scope. +/// If the callback throws, Relay preserves the current emitted payload and records the error for +/// `getLastCallbackError()`. #[napi] pub fn scope_register_llm_sanitize_response_guardrail( + env: Env, scope_uuid: String, name: String, priority: i32, - guardrail: ThreadsafeFunction, + guardrail: JsFunction, ) -> Result<()> { let uuid = uuid::Uuid::parse_str(&scope_uuid) .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; @@ -3079,7 +3191,7 @@ pub fn scope_register_llm_sanitize_response_guardrail( &uuid, &name, priority, - callable::wrap_js_llm_response_fn(guardrail), + callable::wrap_js_llm_response_fn(middleware_json_callback_tsfn(&env, &guardrail)?), ) .map_err(to_napi_err) } @@ -3102,12 +3214,14 @@ pub fn scope_deregister_llm_sanitize_response_guardrail( /// /// The `guardrail` callback receives the LLM request as JSON and must return `null` to allow /// execution or a rejection reason string to block it. Higher `priority` values run first. +/// If the callback throws, the managed call rejects and the protected callback does not run. #[napi] pub fn scope_register_llm_conditional_execution_guardrail( + env: Env, scope_uuid: String, name: String, priority: i32, - guardrail: ThreadsafeFunction, + guardrail: JsFunction, ) -> Result<()> { let uuid = uuid::Uuid::parse_str(&scope_uuid) .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; @@ -3115,7 +3229,7 @@ pub fn scope_register_llm_conditional_execution_guardrail( &uuid, &name, priority, - callable::wrap_js_llm_conditional_fn(guardrail), + callable::wrap_js_llm_conditional_fn(middleware_json_callback_tsfn(&env, &guardrail)?), ) .map_err(to_napi_err) } @@ -3143,8 +3257,10 @@ pub fn scope_deregister_llm_conditional_execution_guardrail( /// The `callable` receives the `LlmRequest` (as JSON) and returns a transformed request. /// If `breakChain` is `true`, no lower-priority intercepts run after this one. /// Higher `priority` values run first. +/// If the callback throws, the managed call rejects and later middleware does not run. #[napi] pub fn scope_register_llm_request_intercept( + env: Env, scope_uuid: String, name: String, priority: i32, @@ -3152,16 +3268,17 @@ pub fn scope_register_llm_request_intercept( #[napi( ts_arg_type = "(args: { name: string; request: Json; annotated: Json | null }) => { request: Json; annotated?: Json | null; pendingMarks?: Array<{ name: string; category?: string | null; categoryProfile?: Json; data?: Json; metadata?: Json }>; optimizationContributions?: Array<{ id?: string; sequence?: number; producer: string; kind: 'input_compression' | 'model_routing' | (string & {}); applied: boolean; model_transition?: { baseline?: { model: string; provider?: string }; effective?: { model: string; provider?: string } }; token_impact?: { baseline?: { prompt_tokens?: number; completion_tokens?: number; cache_read_tokens?: number; cache_write_tokens?: number; total_tokens?: number }; effective?: { prompt_tokens?: number; completion_tokens?: number; cache_read_tokens?: number; cache_write_tokens?: number; total_tokens?: number }; saved?: { prompt_tokens?: number; completion_tokens?: number; cache_read_tokens?: number; cache_write_tokens?: number; total_tokens?: number }; quality?: 'observed' | 'estimated'; estimation_method?: string }; payload_schema?: { name: string; version: string }; payload?: Json; [key: string]: Json | undefined }> }" )] - callable: ThreadsafeFunction, + callable: JsFunction, ) -> Result<()> { let uuid = uuid::Uuid::parse_str(&scope_uuid) .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; + let callback = middleware_json_callback_tsfn(&env, &callable)?; core_registry_api::scope_register_llm_request_intercept( &uuid, &name, priority, break_chain, - callable::wrap_js_llm_request_intercept_fn(callable), + callable::wrap_js_llm_request_intercept_fn(callback), ) .map_err(to_napi_err) } diff --git a/crates/node/src/callable.rs b/crates/node/src/callable.rs index c888bb1a5..0ad24b691 100644 --- a/crates/node/src/callable.rs +++ b/crates/node/src/callable.rs @@ -14,6 +14,7 @@ use std::pin::Pin; use std::sync::Arc; use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode}; +use napi::{Env, JsFunction, JsUnknown, NapiRaw, NapiValue}; use nemo_relay::api::runtime::{ EventSanitizeFn, EventSubscriberFn, LlmConditionalFn, LlmExecutionNextFn, LlmJsonStream, LlmRequestInterceptFn, LlmSanitizeRequestFn, LlmSanitizeResponseFn, LlmStreamExecutionNextFn, @@ -37,7 +38,7 @@ use nemo_relay::error::{FlowError, Result}; use crate::convert::{callback_json, record_callback_error}; use crate::promise_call::{JsonNextFn, JsonStreamNextFn, PromiseAwareFn}; -use crate::types::{EventSanitizeFields, JsEvent, event_sanitize_fields_from_js}; +use crate::types::{EventSanitizeFields, JsEvent, event_sanitize_fields_from_json}; /// JavaScript-facing pending mark DTO. #[derive(Debug, Deserialize, Serialize)] @@ -84,29 +85,101 @@ pub(crate) fn js_pending_marks(marks: Vec) -> Vec, error_prefix: &str) -> Json { - rx.recv().unwrap_or_else(|e| { - record_callback_error(format!("{error_prefix}: {e}")); - Json::Null - }) +#[derive(Deserialize)] +struct MiddlewareCallbackResult { + ok: bool, + #[serde(default)] + value: Json, + #[serde(default)] + error: String, } -fn recv_json_result(rx: std::sync::mpsc::Receiver, error_prefix: &str) -> Result { - rx.recv() - .map_err(|e| FlowError::Internal(format!("{error_prefix}: {e}"))) +/// Wrap a middleware callback so exceptions cross the N-API boundary as data. +/// +/// A raw `ThreadsafeFunction::call_with_return_value` aborts the Node process when +/// a JavaScript callback throws. This wrapper preserves the callback signature and +/// returns a JSON envelope that the Rust middleware adapters can decode safely. +pub(crate) fn safe_middleware_callback(env: &Env, func: &JsFunction) -> napi::Result { + let factory: JsFunction = env.run_script( + r#"((fn) => function __nemo_relay_middleware_wrapper(...args) { + try { + const value = fn(...args); + return { ok: true, value: value === undefined ? null : value }; + } catch (error) { + return { ok: false, error: String(error?.message ?? error) }; + } +})"#, + )?; + let func_unknown = unsafe { JsUnknown::from_raw_unchecked(env.raw(), func.raw()) }; + let wrapper_unknown = factory.call(None, &[func_unknown])?; + Ok(unsafe { wrapper_unknown.cast::() }) +} + +pub(crate) fn unwrap_middleware_result(value: Json, error_prefix: &str) -> Result { + let result: MiddlewareCallbackResult = serde_json::from_value(value).map_err(|error| { + FlowError::Internal(format!( + "{error_prefix}: invalid middleware callback result: {error}" + )) + })?; + if result.ok { + Ok(result.value) + } else { + Err(FlowError::Internal(format!( + "{error_prefix}: {}", + result.error + ))) + } +} + +fn recv_middleware_json_result( + rx: std::sync::mpsc::Receiver, + error_prefix: &str, +) -> Result { + let value = rx + .recv() + .map_err(|error| FlowError::Internal(format!("{error_prefix}: {error}")))?; + unwrap_middleware_result(value, error_prefix) } -fn recv_json_or_value( +fn recv_middleware_json_or_value( rx: std::sync::mpsc::Receiver, error_prefix: &str, fallback: Json, ) -> Json { + match recv_middleware_json_result(rx, error_prefix) { + Ok(value) => value, + Err(error) => { + record_callback_error(error.to_string()); + fallback + } + } +} + +fn recv_middleware_option_string_result( + rx: std::sync::mpsc::Receiver, + error_prefix: &str, +) -> Result> { + match recv_middleware_json_result(rx, error_prefix)? { + Json::Null => Ok(None), + Json::String(value) => Ok(Some(value)), + other => Err(FlowError::Internal(format!( + "{error_prefix}: expected string or null, got {other:?}", + ))), + } +} + +fn recv_json_or_null(rx: std::sync::mpsc::Receiver, error_prefix: &str) -> Json { rx.recv().unwrap_or_else(|e| { record_callback_error(format!("{error_prefix}: {e}")); - fallback + Json::Null }) } +fn recv_json_result(rx: std::sync::mpsc::Receiver, error_prefix: &str) -> Result { + rx.recv() + .map_err(|e| FlowError::Internal(format!("{error_prefix}: {e}"))) +} + fn recv_option_string_result( rx: std::sync::mpsc::Receiver, error_prefix: &str, @@ -120,20 +193,6 @@ fn recv_option_string_result( } } -fn recv_llm_request_or_value( - rx: std::sync::mpsc::Receiver, - error_prefix: &str, - fallback: LlmRequest, -) -> LlmRequest { - let result = recv_json_or_null(rx, error_prefix); - serde_json::from_value(result).unwrap_or_else(|e| { - record_callback_error(format!( - "{error_prefix}: failed to deserialize LlmRequest: {e}" - )); - fallback - }) -} - fn recv_llm_request_result( rx: std::sync::mpsc::Receiver, error_prefix: &str, @@ -154,6 +213,7 @@ pub fn wrap_js_tool_fn( Arc::new(move |name: &str, args: Json| { let func = func.clone(); let name = name.to_string(); + let fallback = args.clone(); let (tx, rx) = std::sync::mpsc::channel(); let status = func.call_with_return_value( (name, args), @@ -167,11 +227,11 @@ pub fn wrap_js_tool_fn( record_callback_error(format!( "nemo_relay: failed to queue JS tool callback: {status:?}" )); - return Json::Null; + return fallback; } // TODO: This closure returns Json (not Result), so we cannot propagate // errors through the type system. Log the error so failures are not silent. - recv_json_or_null(rx, "nemo_relay: JS tool callback failed") + recv_middleware_json_or_value(rx, "nemo_relay: JS tool callback failed", fallback) }) } @@ -198,7 +258,7 @@ pub fn wrap_js_tool_conditional_fn( "failed to queue JS tool conditional callback: {status:?}", ))); } - recv_option_string_result(rx, "JS tool conditional callback failed") + recv_middleware_option_string_result(rx, "JS tool conditional callback failed") }) } @@ -224,7 +284,7 @@ pub fn wrap_js_tool_request_intercept_fn( "failed to queue JS tool callback: {status:?}", ))); } - recv_json_result(rx, "JS tool callback failed") + recv_middleware_json_result(rx, "JS tool callback failed") }) } @@ -296,7 +356,8 @@ pub fn wrap_js_llm_request_intercept_fn( "failed to queue JS LLM request intercept callback: {status:?}", ))); } - let result = recv_json_result(rx, "JS LLM request intercept callback failed")?; + let result = + recv_middleware_json_result(rx, "JS LLM request intercept callback failed")?; #[derive(Deserialize)] #[serde(rename_all = "camelCase")] @@ -348,11 +409,17 @@ pub fn wrap_js_llm_sanitize_request_fn( } // TODO: This closure returns LlmRequest (not Result), so we cannot propagate // errors through the type system. Log the error so failures are not silent. - recv_llm_request_or_value( + let result = recv_middleware_json_or_value( rx, "nemo_relay: JS LLM sanitize request callback failed", - request, - ) + serde_json::to_value(&request).unwrap_or(Json::Null), + ); + serde_json::from_value(result).unwrap_or_else(|error| { + record_callback_error(format!( + "nemo_relay: JS LLM sanitize request callback failed: failed to deserialize LlmRequest: {error}" + )); + request + }) }) } @@ -380,7 +447,7 @@ pub fn wrap_js_llm_response_fn( } // TODO: This closure returns Json (not Result), so we cannot propagate // errors through the type system. Log the error and fall back to original response. - recv_json_or_value(rx, "nemo_relay: JS LLM response callback failed", response) + recv_middleware_json_or_value(rx, "nemo_relay: JS LLM response callback failed", response) }) } @@ -406,7 +473,7 @@ pub fn wrap_js_llm_conditional_fn( "failed to queue JS LLM conditional callback: {status:?}", ))); } - recv_option_string_result(rx, "JS LLM conditional callback failed") + recv_middleware_option_string_result(rx, "JS LLM conditional callback failed") }) } @@ -547,8 +614,8 @@ pub fn wrap_js_event_sanitize_fn( serde_json::to_value(js_fields).unwrap_or(Json::Null), ), ThreadsafeFunctionCallMode::Blocking, - move |value: napi::JsUnknown| { - let _ = tx.send(event_sanitize_fields_from_js(value)); + move |value: Option| { + let _ = tx.send(callback_json(value)); Ok(()) }, ); @@ -558,30 +625,36 @@ pub fn wrap_js_event_sanitize_fn( )); return fields.clone(); } - rx.recv() - .map_err(|error| { - record_callback_error(format!( - "nemo_relay: JS event sanitizer callback failed: {error}" - )); - }) - .ok() - .and_then(|result| result.ok()) - .and_then(|result| { - let category_profile = result - .category_profile - .map(serde_json::from_value) - .transpose() - .ok()?; - Some(CoreEventSanitizeFields { - data: result.data, - category_profile, - metadata: result.metadata, - }) + let sanitized = (|| -> Result<_> { + let result = + recv_middleware_json_result(rx, "nemo_relay: JS event sanitizer callback failed")?; + let result = event_sanitize_fields_from_json(result).map_err(|error| { + FlowError::Internal(format!( + "nemo_relay: invalid JS event sanitizer result: {error}" + )) + })?; + let category_profile = result + .category_profile + .map(serde_json::from_value) + .transpose() + .map_err(|error| { + FlowError::Internal(format!( + "nemo_relay: invalid JS event sanitizer result: {error}" + )) + })?; + Ok(CoreEventSanitizeFields { + data: result.data, + category_profile, + metadata: result.metadata, }) - .unwrap_or_else(|| { - record_callback_error("nemo_relay: invalid JS event sanitizer result".to_string()); + })(); + match sanitized { + Ok(sanitized) => sanitized, + Err(error) => { + record_callback_error(error.to_string()); fields.clone() - }) + } + } }) } diff --git a/crates/node/src/types/mod.rs b/crates/node/src/types/mod.rs index 6f5f96fe6..63c594f30 100644 --- a/crates/node/src/types/mod.rs +++ b/crates/node/src/types/mod.rs @@ -7,7 +7,6 @@ //! and attribute constants that are exposed to JavaScript/TypeScript consumers. //! Doc comments on `#[napi]` items are emitted into the generated `index.d.ts`. -use napi::{JsObject, JsUnknown}; use napi_derive::napi; use nemo_relay::api::runtime::{ScopeStackHandle, create_scope_stack}; use serde::{Deserialize, Serialize}; @@ -310,43 +309,23 @@ pub struct EventSanitizeFields { pub metadata: Option, } -pub(crate) fn event_sanitize_fields_from_js(value: JsUnknown) -> napi::Result { - let object = match value.get_type()? { - napi::ValueType::Object => JsObject::try_from(value)?, - _ => { - return Err(napi::Error::from_reason( - "event sanitizer must return an object", - )); - } +pub(crate) fn event_sanitize_fields_from_json( + value: Json, +) -> serde_json::Result { + let Some(object) = value.as_object() else { + return Err(serde_json::Error::io(std::io::Error::other( + "event sanitizer must return an object", + ))); }; - if object.is_array()? || object.has_named_property("then")? { - return Err(napi::Error::from_reason( - "event sanitizer must return a non-thenable object", - )); - } - - let mut has_field = false; - for field in ["data", "categoryProfile", "metadata"] { - has_field |= object.has_own_property(field)?; - } - if !has_field { - return Err(napi::Error::from_reason( + if !["data", "categoryProfile", "metadata"] + .iter() + .any(|field| object.contains_key(*field)) + { + return Err(serde_json::Error::io(std::io::Error::other( "event sanitizer must return at least one sanitizer field", - )); + ))); } - - let field = |name| { - if object.has_own_property(name)? { - object.get(name) - } else { - Ok(None) - } - }; - Ok(EventSanitizeFields { - data: field("data")?, - category_profile: field("categoryProfile")?, - metadata: field("metadata")?, - }) + serde_json::from_value(value) } // --------------------------------------------------------------------------- diff --git a/crates/node/tests/adaptive_tests.mjs b/crates/node/tests/adaptive_tests.mjs index 960ecb2c6..bf2c4b090 100644 --- a/crates/node/tests/adaptive_tests.mjs +++ b/crates/node/tests/adaptive_tests.mjs @@ -6,6 +6,7 @@ import assert from 'node:assert/strict'; import { createRequire } from 'node:module'; const require = createRequire(import.meta.url); +const lib = require('../index.js'); const plugin = require('../plugin.js'); const adaptive = require('../adaptive.js'); @@ -166,6 +167,39 @@ describe('core plugins', () => { plugin.deregister(pluginKind); } }); + + it('turns plugin request-intercept throws into catchable errors', async () => { + const pluginKind = `node.test.request-throw.${Date.now()}`; + plugin.register(pluginKind, { + register(_config, context) { + context.registerToolRequestIntercept('throwingRequest', 10, false, () => { + throw new Error('plugin request intercept boom'); + }); + }, + }); + + try { + const report = await plugin.initialize({ + version: 1, + components: [ + adaptive.ComponentSpec({ + version: 1, + state: { backend: adaptive.inMemoryBackend() }, + adaptive_hints: adaptive.adaptiveHintsConfig(), + }), + plugin.ComponentSpec(pluginKind, {}), + ], + }); + assert.deepEqual(report.diagnostics, []); + await assert.rejects( + () => lib.toolCallExecute('plugin_request_throw', {}, () => ({ should_not: 'run' })), + /plugin request intercept boom/i, + ); + } finally { + plugin.clear(); + plugin.deregister(pluginKind); + } + }); }); describe('adaptive helpers', () => { diff --git a/crates/node/tests/callback_error_tests.mjs b/crates/node/tests/callback_error_tests.mjs index 80f6e3995..94130142b 100644 --- a/crates/node/tests/callback_error_tests.mjs +++ b/crates/node/tests/callback_error_tests.mjs @@ -67,17 +67,18 @@ describe('callback error helpers', () => { } }); - it('closed tool callbacks fall back to null and record the queue failure', () => { + it('closed tool sanitize callbacks preserve the original payload and record the queue failure', () => { + const args = { + value: 1, + }; const result = __testClosedToolCallback( () => ({ ok: true, }), 'closed_tool', - { - value: 1, - }, + args, ); - assert.equal(result, null); + assert.deepEqual(result, args); assert.match(getLastCallbackError() ?? '', /failed to queue JS tool callback/i); clearLastCallbackError(); }); diff --git a/crates/node/tests/event_sanitizers_tests.mjs b/crates/node/tests/event_sanitizers_tests.mjs index e4f776c9c..22ab16891 100644 --- a/crates/node/tests/event_sanitizers_tests.mjs +++ b/crates/node/tests/event_sanitizers_tests.mjs @@ -164,6 +164,29 @@ describe('event sanitizer registries', () => { } }); + it('fails open with the original payload when a thread-safe sanitizer throws', async () => { + const events = capture('node-event-sanitize-background-throw-sub'); + lib.clearLastCallbackError(); + lib.registerScopeSanitizeStartGuardrail('node-background-throw', 0, () => { + throw new Error('background sanitizer boom'); + }); + try { + await lib.toolCallExecute('background-throw-tool', { kept: true }, async (args) => args); + lib.flushSubscribers(); + await waitFor(events, 2); + const start = events.find( + (event) => + event.kind === 'scope' && event.name === 'background-throw-tool' && event.scope_category === 'start', + ); + assert.deepEqual(start.data, { kept: true }); + assert.match(lib.getLastCallbackError() ?? '', /background sanitizer boom/i); + } finally { + lib.deregisterScopeSanitizeStartGuardrail('node-background-throw'); + lib.deregisterSubscriber('node-event-sanitize-background-throw-sub'); + lib.clearLastCallbackError(); + } + }); + it('inherits and cleans up scope-local mark sanitizers', async () => { const events = capture('node-event-sanitize-local-sub'); const owner = lib.pushScope('owner', lib.ScopeType.Agent); @@ -219,4 +242,30 @@ describe('event sanitizer registries', () => { assert.deepEqual(marks.configured.data, { plugin: true }); assert.deepEqual(marks.cleared.data, { raw: true }); }); + + it('fails open when a plugin-owned sanitizer throws', async () => { + const kind = `node.test.event-sanitize-throw.${Date.now()}`; + const events = capture('node-event-sanitize-plugin-throw-sub'); + plugin.register(kind, { + register(_config, context) { + context.registerMarkSanitizeGuardrail('mark', 0, () => { + throw new Error('plugin sanitizer boom'); + }); + }, + }); + lib.clearLastCallbackError(); + try { + await plugin.initialize({ version: 1, components: [plugin.ComponentSpec(kind)] }); + lib.event('plugin-throw', null, { raw: true }); + lib.flushSubscribers(); + await waitFor(events, 1); + assert.deepEqual(events.at(-1).data, { raw: true }); + assert.match(lib.getLastCallbackError() ?? '', /plugin sanitizer boom/i); + } finally { + plugin.clear(); + plugin.deregister(kind); + lib.deregisterSubscriber('node-event-sanitize-plugin-throw-sub'); + lib.clearLastCallbackError(); + } + }); }); diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index f48144941..e0f051a87 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/crates/node/tests/llm_tests.mjs @@ -491,6 +491,34 @@ describe('LLM guardrails', () => { } }); + it('conditional guardrail throws a catchable error without terminating Node', async () => { + registerLlmConditionalExecutionGuardrail('node_llm_cond_throw', 10, () => { + throw new Error('llm guardrail boom'); + }); + try { + await assert.rejects( + () => + llmCallExecute('llm_cond_throw', makeNative(), () => ({ should_not: 'run' }), null, null, null, null, null), + /llm guardrail boom/i, + ); + deregisterLlmConditionalExecutionGuardrail('node_llm_cond_throw'); + + const result = await llmCallExecute( + 'llm_after_guardrail_throw', + makeNative(), + () => ({ ok: true }), + null, + null, + null, + null, + null, + ); + assert.deepEqual(result, { ok: true }); + } finally { + deregisterLlmConditionalExecutionGuardrail('node_llm_cond_throw'); + } + }); + it('conditional guardrail (block)', () => { registerLlmConditionalExecutionGuardrail('node_llm_block', 10, (request) => 'blocked'); deregisterLlmConditionalExecutionGuardrail('node_llm_block'); @@ -576,6 +604,33 @@ describe('LLM intercepts', () => { deregisterLlmRequestIntercept('node_llm_req_mod'); }); + it('request intercept throws a catchable error without terminating Node', async () => { + registerLlmRequestIntercept('node_llm_req_throw', 10, false, () => { + throw new Error('llm request intercept boom'); + }); + try { + await assert.rejects( + () => + llmCallExecute('llm_req_throw', makeNative(), () => ({ should_not: 'run' }), null, null, null, null, null), + /llm request intercept boom/i, + ); + deregisterLlmRequestIntercept('node_llm_req_throw'); + const result = await llmCallExecute( + 'llm_after_request_intercept_throw', + makeNative(), + () => ({ ok: true }), + null, + null, + null, + null, + null, + ); + assert.deepEqual(result, { ok: true }); + } finally { + deregisterLlmRequestIntercept('node_llm_req_throw'); + } + }); + it('request intercept rejects malformed return values', async () => { registerLlmRequestIntercept('node_llm_req_bad', 10, false, () => null); try { diff --git a/crates/node/tests/scope_local_tests.mjs b/crates/node/tests/scope_local_tests.mjs index 14dc49dd0..e478a73b1 100644 --- a/crates/node/tests/scope_local_tests.mjs +++ b/crates/node/tests/scope_local_tests.mjs @@ -205,6 +205,22 @@ describe('Scope-local guardrail registration and execution', () => { popScope(scope); }); + it('scope-local conditional guardrail throws a catchable error', async () => { + const scope = pushScope('sl_guard_throw', ScopeType.Agent, null, null); + scopeRegisterToolConditionalExecutionGuardrail(scope.uuid, 'sl_throw_1', 10, () => { + throw new Error('scope guardrail boom'); + }); + try { + await assert.rejects( + () => toolCallExecute('sl_throw_tool', {}, () => ({ should_not: 'run' }), null, null, null, null), + /scope guardrail boom/i, + ); + } finally { + scopeDeregisterToolConditionalExecutionGuardrail(scope.uuid, 'sl_throw_1'); + popScope(scope); + } + }); + it('duplicate scope-local guardrail name fails', () => { const scope = pushScope('sl_guard_dup', ScopeType.Agent, null, null); scopeRegisterToolSanitizeRequestGuardrail(scope.uuid, 'sl_dup_guard', 10, (n, a) => a); @@ -695,6 +711,22 @@ describe('Priority merge of global and scope-local middleware', () => { lib.deregisterToolRequestIntercept('sl_merge_global_int'); }); + it('scope-local tool request intercept throws a catchable error', async () => { + const scope = pushScope('sl_tool_request_throw_scope', ScopeType.Agent, null, null); + scopeRegisterToolRequestIntercept(scope.uuid, 'sl_tool_request_throw', 10, false, () => { + throw new Error('scope tool request intercept boom'); + }); + try { + await assert.rejects( + () => toolCallExecute('sl_tool_request_throw_call', {}, () => ({ should_not: 'run' }), null, null, null, null), + /scope tool request intercept boom/i, + ); + } finally { + scopeDeregisterToolRequestIntercept(scope.uuid, 'sl_tool_request_throw'); + popScope(scope); + } + }); + it('scope-local execution intercept and global intercept merge', async () => { lib.registerToolExecutionIntercept('sl_merge_global_exec', 5, async (args, next) => { const result = await next({ diff --git a/crates/node/tests/tools_tests.mjs b/crates/node/tests/tools_tests.mjs index 6cb9fdc0b..afb79eb02 100644 --- a/crates/node/tests/tools_tests.mjs +++ b/crates/node/tests/tools_tests.mjs @@ -27,6 +27,8 @@ const { deregisterToolRequestIntercept, registerToolExecutionIntercept, deregisterToolExecutionIntercept, + clearLastCallbackError, + getLastCallbackError, registerSubscriber, deregisterSubscriber, flushSubscribers, @@ -520,6 +522,72 @@ describe('Tool guardrails', () => { } }); + it('conditional guardrail throws a catchable error without terminating Node', async () => { + let executed = false; + registerToolConditionalExecutionGuardrail('node_tool_cond_throw', 10, () => { + throw new Error('guardrail boom'); + }); + try { + await assert.rejects( + () => + toolCallExecute( + 'tool_cond_throw', + {}, + () => { + executed = true; + return {}; + }, + null, + null, + null, + null, + ), + /guardrail boom/i, + ); + assert.equal(executed, false); + deregisterToolConditionalExecutionGuardrail('node_tool_cond_throw'); + + const result = await toolCallExecute( + 'tool_after_guardrail_throw', + {}, + () => ({ ok: true }), + null, + null, + null, + null, + ); + assert.deepEqual(result, { ok: true }); + } finally { + deregisterToolConditionalExecutionGuardrail('node_tool_cond_throw'); + } + }); + + it('sanitize guardrail failures preserve the observable payload', async () => { + clearLastCallbackError(); + const events = []; + registerSubscriber('node_tool_san_throw_sub', (event) => events.push(event)); + registerToolSanitizeRequestGuardrail('node_tool_san_throw', 10, () => { + throw new Error('sanitize boom'); + }); + try { + await toolCallExecute('tool_sanitize_throw', { original: true }, (args) => args, null, null, null, null); + await waitForSubscriberCallbacks(() => events.some((event) => event.name === 'tool_sanitize_throw')); + const start = events.find( + (event) => + event.name === 'tool_sanitize_throw' && + event.kind === 'scope' && + event.category === 'tool' && + event.scope_category === 'start', + ); + assert.deepEqual(start.data, { original: true }); + assert.match(getLastCallbackError() ?? '', /sanitize boom/i); + } finally { + deregisterToolSanitizeRequestGuardrail('node_tool_san_throw'); + deregisterSubscriber('node_tool_san_throw_sub'); + clearLastCallbackError(); + } + }); + it('conditional guardrail (block)', () => { registerToolConditionalExecutionGuardrail('node_tool_block', 10, (name, args) => 'blocked'); deregisterToolConditionalExecutionGuardrail('node_tool_block'); @@ -606,6 +674,31 @@ describe('Tool intercepts', () => { deregisterToolRequestIntercept('node_tool_req_mod'); }); + it('request intercept throws a catchable error without terminating Node', async () => { + registerToolRequestIntercept('node_tool_req_throw', 10, false, () => { + throw new Error('tool request intercept boom'); + }); + try { + await assert.rejects( + () => toolCallExecute('tool_req_throw', {}, () => ({ should_not: 'run' }), null, null, null, null), + /tool request intercept boom/i, + ); + deregisterToolRequestIntercept('node_tool_req_throw'); + const result = await toolCallExecute( + 'tool_after_request_intercept_throw', + {}, + () => ({ ok: true }), + null, + null, + null, + null, + ); + assert.deepEqual(result, { ok: true }); + } finally { + deregisterToolRequestIntercept('node_tool_req_throw'); + } + }); + it('request intercept can return null JSON', async () => { registerToolRequestIntercept('node_tool_req_bad', 10, false, () => null); try { diff --git a/docs/instrument-applications/advanced-guide.mdx b/docs/instrument-applications/advanced-guide.mdx index 8bf0517ca..87c41c380 100644 --- a/docs/instrument-applications/advanced-guide.mdx +++ b/docs/instrument-applications/advanced-guide.mdx @@ -192,6 +192,14 @@ NeMo Relay exposes the same core middleware families for tools and LLMs: Sanitize guardrails affect only the payload recorded on emitted events. Request intercepts affect the real request that reaches the tool or provider. Execution intercepts wrap the callback itself and are only available when the invocation uses managed execution. +### Node.js Callback Failures + +Node.js guardrail and request-intercept callbacks are synchronous. A thrown +conditional-execution guardrail or request intercept rejects the managed call +before protected execution or later middleware runs. A thrown sanitize guardrail +or event sanitizer fails open, preserving the current emitted payload and +recording the error for `getLastCallbackError()`. + Scope-local variants are available through `nemo_relay.scope_local.register_*`, Node.js `scopeRegister*` helpers, and Rust `scope_register_*` functions. ## Validate the Middleware diff --git a/docs/reference/event-sanitizers.mdx b/docs/reference/event-sanitizers.mdx index feadbd37b..8ded00de5 100644 --- a/docs/reference/event-sanitizers.mdx +++ b/docs/reference/event-sanitizers.mdx @@ -49,8 +49,9 @@ semantic category, attributes, semantic input and output meaning, or schemas. Registries run in priority order. Lower priorities run first, and each callback receives the fields returned by the callback before it. Invalid -binding callback results fail open and preserve the current fields. Node.js -records these failures for `getLastCallbackError()`. +binding callback results fail open and preserve the current fields. In Node.js, +a synchronous sanitizer callback that throws also fails open; Relay records the +error for `getLastCallbackError()`. ## Registration Lifetimes From 47fe5590982b5e3682f5365e44a1e230265c7b99 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Sat, 18 Jul 2026 19:26:29 -0400 Subject: [PATCH 2/2] test: cover Node middleware sanitizer failures Signed-off-by: Will Killian --- crates/node/src/callable.rs | 6 ++- crates/node/tests/llm_tests.mjs | 89 +++++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 1 deletion(-) diff --git a/crates/node/src/callable.rs b/crates/node/src/callable.rs index 0ad24b691..1cb21a4b2 100644 --- a/crates/node/src/callable.rs +++ b/crates/node/src/callable.rs @@ -106,7 +106,11 @@ pub(crate) fn safe_middleware_callback(env: &Env, func: &JsFunction) -> napi::Re const value = fn(...args); return { ok: true, value: value === undefined ? null : value }; } catch (error) { - return { ok: false, error: String(error?.message ?? error) }; + let message = 'JavaScript callback threw'; + try { + message = String(error?.message ?? error); + } catch {} + return { ok: false, error: message }; } })"#, )?; diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index e0f051a87..2500c0a24 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/crates/node/tests/llm_tests.mjs @@ -34,6 +34,8 @@ const { registerSubscriber, deregisterSubscriber, flushSubscribers, + clearLastCallbackError, + getLastCallbackError, ScopeType, } = lib; @@ -381,6 +383,53 @@ describe('LLM guardrails', () => { } }); + it('sanitize request guardrail failures preserve the payload and remain usable', async () => { + const events = []; + clearLastCallbackError(); + registerSubscriber('node_llm_san_req_throw_sub', (event) => events.push(event)); + registerLlmSanitizeRequestGuardrail('node_llm_san_req_throw', 10, () => { + throw { + get message() { + throw new Error('message getter boom'); + }, + toString() { + throw new Error('string conversion boom'); + }, + }; + }); + try { + const request = makeNative(); + await llmCallExecute('llm_san_req_throw', request, () => ({ ok: true }), null, null, null, null, null); + await flushSubscriberCallbacks(); + const start = events.find( + (event) => + event.name === 'llm_san_req_throw' && + event.kind === 'scope' && + event.category === 'llm' && + event.scope_category === 'start', + ); + assert.deepEqual(start.data, request); + assert.match(getLastCallbackError() ?? '', /JavaScript callback threw/i); + + deregisterLlmSanitizeRequestGuardrail('node_llm_san_req_throw'); + const result = await llmCallExecute( + 'llm_after_san_req_throw', + makeNative(), + () => ({ ok: true }), + null, + null, + null, + null, + null, + ); + assert.deepEqual(result, { ok: true }); + } finally { + deregisterLlmSanitizeRequestGuardrail('node_llm_san_req_throw'); + deregisterSubscriber('node_llm_san_req_throw_sub'); + clearLastCallbackError(); + } + }); + it('conditional guardrail rejects non-string return values', async () => { registerLlmConditionalExecutionGuardrail('node_llm_cond_non_string', 10, () => ({ blocked: true, @@ -463,6 +512,46 @@ describe('LLM guardrails', () => { } }); + it('sanitize response guardrail failures preserve the payload and remain usable', async () => { + const events = []; + clearLastCallbackError(); + registerSubscriber('node_llm_san_resp_throw_sub', (event) => events.push(event)); + registerLlmSanitizeResponseGuardrail('node_llm_san_resp_throw', 10, () => { + throw new Error('response sanitizer boom'); + }); + try { + const response = { ok: true }; + await llmCallExecute('llm_san_resp_throw', makeNative(), () => response, null, null, null, null, null); + await flushSubscriberCallbacks(); + const end = events.find( + (event) => + event.name === 'llm_san_resp_throw' && + event.kind === 'scope' && + event.category === 'llm' && + event.scope_category === 'end', + ); + assert.deepEqual(end.data, response); + assert.match(getLastCallbackError() ?? '', /response sanitizer boom/i); + + deregisterLlmSanitizeResponseGuardrail('node_llm_san_resp_throw'); + const result = await llmCallExecute( + 'llm_after_san_resp_throw', + makeNative(), + () => ({ ok: true }), + null, + null, + null, + null, + null, + ); + assert.deepEqual(result, { ok: true }); + } finally { + deregisterLlmSanitizeResponseGuardrail('node_llm_san_resp_throw'); + deregisterSubscriber('node_llm_san_resp_throw_sub'); + clearLastCallbackError(); + } + }); + it('conditional guardrail (allow)', () => { registerLlmConditionalExecutionGuardrail('node_llm_cond', 10, (request) => null); deregisterLlmConditionalExecutionGuardrail('node_llm_cond');