diff --git a/crates/core/src/api/llm.rs b/crates/core/src/api/llm.rs index f0333bb91..2c2877944 100644 --- a/crates/core/src/api/llm.rs +++ b/crates/core/src/api/llm.rs @@ -706,6 +706,9 @@ fn emit_optimization_marks_with( /// This emits an LLM-start event after applying sanitize-request guardrails to /// the payload recorded for observability. /// +/// If a sanitizer errors or panics, Relay omits the payload and request +/// annotation and does not run remaining sanitizers. +/// /// # Parameters /// - `name`: Logical provider or model family name recorded on the span. /// - `request`: Raw [`LlmRequest`] associated with the span. @@ -927,12 +930,14 @@ async fn build_llm_end_payload( /// # Errors /// Returns an error when the runtime owner check fails or internal state cannot /// be read safely. Dispatcher submission failures are logged because -/// observability publication is best effort. Sanitizer and response-codec errors -/// discovered during queued publication are also logged and fail open. +/// observability publication is best effort. Sanitizer errors discovered during +/// queued publication are logged and fail closed by omitting the governed payload. +/// Response-codec errors retain their documented fallback behavior. /// /// # Notes /// Sanitize-response guardrails affect only the emitted end-event payload, not -/// the caller-owned `response` value. +/// the caller-owned `response` value. If a sanitizer errors or panics, Relay +/// omits the payload and response annotation and does not run remaining sanitizers. pub fn llm_call_end(params: LlmCallEndParams<'_>) -> Result<()> { ensure_runtime_owner()?; let scope_stack = params.handle.captured_scope_stack().clone(); diff --git a/crates/core/src/api/runtime/state.rs b/crates/core/src/api/runtime/state.rs index 15db15dc2..fa2da2cee 100644 --- a/crates/core/src/api/runtime/state.rs +++ b/crates/core/src/api/runtime/state.rs @@ -20,8 +20,9 @@ use std::task::{Context, Poll}; use futures_util::{FutureExt, Stream}; use crate::api::event::{ - BaseEvent, CategoryProfile, Event, EventCategory, MarkEvent, ScopeCategory, ScopeEvent, - llm_attributes_to_strings, scope_attributes_to_strings, tool_attributes_to_strings, + BaseEvent, CategoryProfile, Event, EventCategory, EventSanitizeFields, MarkEvent, + ScopeCategory, ScopeEvent, llm_attributes_to_strings, scope_attributes_to_strings, + tool_attributes_to_strings, }; use crate::api::llm::{CreateLlmHandleParams, EndLlmHandleParams}; use crate::api::llm::{LlmHandle, LlmRequest}; @@ -810,20 +811,28 @@ impl NemoRelayContextState { event = Arc::try_unwrap(context).unwrap_or_else(|context| (*context).clone()); match outcome { Ok(Ok(fields)) => event.apply_sanitize_fields(fields), - Ok(Err(error)) => log::error!( - target: "nemo_relay.runtime", - event = "event_sanitizer_failed", - sanitizer = entry.name.as_str(), - event_name = event.name(); - "Event sanitizer failed; preserving the last valid event snapshot: {error}" - ), - Err(_) => log::error!( - target: "nemo_relay.runtime", - event = "event_sanitizer_panicked", - sanitizer = entry.name.as_str(), - event_name = event.name(); - "Event sanitizer panicked; publishing the latest valid event snapshot" - ), + Ok(Err(_error)) => { + log::error!( + target: "nemo_relay.runtime", + event = "event_sanitizer_failed", + sanitizer = entry.name.as_str(), + event_name = event.name(); + "Event sanitizer failed; clearing observability fields" + ); + event.apply_sanitize_fields(EventSanitizeFields::default()); + break; + } + Err(_) => { + log::error!( + target: "nemo_relay.runtime", + event = "event_sanitizer_panicked", + sanitizer = entry.name.as_str(), + event_name = event.name(); + "Event sanitizer panicked; clearing observability fields" + ); + event.apply_sanitize_fields(EventSanitizeFields::default()); + break; + } } } event @@ -856,36 +865,38 @@ impl NemoRelayContextState { /// - `entries`: Sanitizer snapshots to evaluate. /// /// # Returns - /// The sanitized JSON payload after every provided guardrail has run. + /// The sanitized JSON payload after every provided guardrail has run, or + /// `None` when a sanitizer failure omits the payload. pub(crate) async fn tool_sanitize_request_snapshot_chain( name: &str, args: Json, entries: &[Guardrail], - ) -> Json { - let mut value = args; + ) -> Option { + let mut value = Some(args); for entry in entries { - let callback = Arc::clone(&entry.payload); - let callback_name = name.to_string(); - let current = value.clone(); - match AssertUnwindSafe(async move { callback(callback_name, current).await }) - .catch_unwind() - .await - { - Ok(Ok(next)) => value = next, - Ok(Err(error)) => log::error!( - target: "nemo_relay.runtime", - event = "tool_request_sanitizer_failed", - sanitizer = entry.name.as_str(), - tool_name = name; - "Tool request sanitizer failed; preserving the last valid payload: {error}" - ), - Err(_) => log::error!( - target: "nemo_relay.runtime", - event = "tool_request_sanitizer_panicked", - sanitizer = entry.name.as_str(), - tool_name = name; - "Tool request sanitizer panicked; preserving the last valid payload" - ), + if let Some(current) = value.take() { + let callback = Arc::clone(&entry.payload); + let callback_name = name.to_string(); + match AssertUnwindSafe(async move { callback(callback_name, current).await }) + .catch_unwind() + .await + { + Ok(Ok(next)) => value = Some(next), + Ok(Err(_error)) => log::error!( + target: "nemo_relay.runtime", + event = "tool_request_sanitizer_failed", + sanitizer = entry.name.as_str(), + tool_name = name; + "Tool request sanitizer failed; omitting the observability payload" + ), + Err(_) => log::error!( + target: "nemo_relay.runtime", + event = "tool_request_sanitizer_panicked", + sanitizer = entry.name.as_str(), + tool_name = name; + "Tool request sanitizer panicked; omitting the observability payload" + ), + } } } value @@ -918,36 +929,38 @@ impl NemoRelayContextState { /// - `entries`: Sanitizer snapshots to evaluate. /// /// # Returns - /// The sanitized JSON payload after every provided guardrail has run. + /// The sanitized JSON payload after every provided guardrail has run, or + /// `None` when a sanitizer failure omits the payload. pub(crate) async fn tool_sanitize_response_snapshot_chain( name: &str, result: Json, entries: &[Guardrail], - ) -> Json { - let mut value = result; + ) -> Option { + let mut value = Some(result); for entry in entries { - let callback = Arc::clone(&entry.payload); - let callback_name = name.to_string(); - let current = value.clone(); - match AssertUnwindSafe(async move { callback(callback_name, current).await }) - .catch_unwind() - .await - { - Ok(Ok(next)) => value = next, - Ok(Err(error)) => log::error!( - target: "nemo_relay.runtime", - event = "tool_response_sanitizer_failed", - sanitizer = entry.name.as_str(), - tool_name = name; - "Tool response sanitizer failed; preserving the last valid payload: {error}" - ), - Err(_) => log::error!( - target: "nemo_relay.runtime", - event = "tool_response_sanitizer_panicked", - sanitizer = entry.name.as_str(), - tool_name = name; - "Tool response sanitizer panicked; preserving the last valid payload" - ), + if let Some(current) = value.take() { + let callback = Arc::clone(&entry.payload); + let callback_name = name.to_string(); + match AssertUnwindSafe(async move { callback(callback_name, current).await }) + .catch_unwind() + .await + { + Ok(Ok(next)) => value = Some(next), + Ok(Err(_error)) => log::error!( + target: "nemo_relay.runtime", + event = "tool_response_sanitizer_failed", + sanitizer = entry.name.as_str(), + tool_name = name; + "Tool response sanitizer failed; omitting the observability payload" + ), + Err(_) => log::error!( + target: "nemo_relay.runtime", + event = "tool_response_sanitizer_panicked", + sanitizer = entry.name.as_str(), + tool_name = name; + "Tool response sanitizer panicked; omitting the observability payload" + ), + } } } value @@ -1226,7 +1239,8 @@ impl NemoRelayContextState { /// - `entries`: Sanitizer snapshots to evaluate. /// /// # Returns - /// The sanitized [`LlmRequest`] after every provided guardrail has run. + /// The sanitized [`LlmRequest`] after every provided guardrail has run, or + /// `None` when a sanitizer errors or panics. pub(crate) async fn llm_sanitize_request_snapshot_chain( request: LlmRequest, context: LlmSanitizeRequestContext, @@ -1245,24 +1259,21 @@ impl NemoRelayContextState { .await { Ok(Ok(next)) => value = next, - Ok(Err(error)) => { + Ok(Err(_error)) => { log::error!( target: "nemo_relay.runtime", event = "llm_request_sanitizer_failed", - sanitizer = entry.name.as_str(), - preserved_value = "last_valid_request"; - "LLM request sanitizer failed; preserving the last valid request: {error}" + sanitizer = entry.name.as_str(); + "LLM request sanitizer failed; omitting the observability payload" ); - value = Some(current); } Err(_) => { log::error!( target: "nemo_relay.runtime", event = "llm_request_sanitizer_panicked", sanitizer = entry.name.as_str(); - "LLM request sanitizer panicked; preserving the last valid request" + "LLM request sanitizer panicked; omitting the observability payload" ); - value = Some(current); } } } @@ -1296,7 +1307,8 @@ impl NemoRelayContextState { /// - `entries`: Sanitizer snapshots to evaluate. /// /// # Returns - /// The sanitized response payload after every provided guardrail has run. + /// The sanitized response payload after every provided guardrail has run, + /// or `None` when a sanitizer errors or panics. pub(crate) async fn llm_sanitize_response_snapshot_chain( response: Json, context: LlmSanitizeResponseContext, @@ -1315,24 +1327,21 @@ impl NemoRelayContextState { .await { Ok(Ok(next)) => value = next, - Ok(Err(error)) => { + Ok(Err(_error)) => { log::error!( target: "nemo_relay.runtime", event = "llm_response_sanitizer_failed", - sanitizer = entry.name.as_str(), - preserved_value = "last_valid_response"; - "LLM response sanitizer failed; preserving the last valid response: {error}" + sanitizer = entry.name.as_str(); + "LLM response sanitizer failed; omitting the observability payload" ); - value = Some(current); } Err(_) => { log::error!( target: "nemo_relay.runtime", event = "llm_response_sanitizer_panicked", sanitizer = entry.name.as_str(); - "LLM response sanitizer panicked; preserving the last valid response" + "LLM response sanitizer panicked; omitting the observability payload" ); - value = Some(current); } } } diff --git a/crates/core/src/api/runtime/subscriber_dispatcher.rs b/crates/core/src/api/runtime/subscriber_dispatcher.rs index 035831900..e4255ffa7 100644 --- a/crates/core/src/api/runtime/subscriber_dispatcher.rs +++ b/crates/core/src/api/runtime/subscriber_dispatcher.rs @@ -3,7 +3,7 @@ //! Asynchronous subscriber delivery for native targets. -use crate::api::event::Event; +use crate::api::event::{Event, EventSanitizeFields}; use crate::api::registry::Guardrail; use crate::api::runtime::{ EventSanitizeFn, EventSubscriberFn, NemoRelayContextState, ScopeStackHandle, @@ -967,8 +967,8 @@ mod native { /// Apply a transform and sanitizers on the dispatcher thread. A transform /// failure drops the event because it may be responsible for inserting the - /// sanitized payload. A sanitizer failure retains the transformed snapshot - /// and continues publication (fail open). + /// sanitized payload. A sanitizer failure clears mutable observability + /// fields before publication. pub(super) fn sanitize_event_snapshot( event: Event, transform: Option, @@ -1037,10 +1037,12 @@ mod native { } log::error!( target: "nemo_relay.runtime", - event = "event_sanitizer_fail_open"; - "Publishing the transformed event snapshot because event sanitizers could not run" + event = "event_sanitizer_runtime_failed"; + "Event sanitizers could not run; clearing observability fields before publication" ); - return (Some(transformed), nested_publications); + let mut cleared = transformed; + cleared.apply_sanitize_fields(EventSanitizeFields::default()); + return (Some(cleared), nested_publications); } }; let fallback = transformed.clone(); @@ -1057,9 +1059,11 @@ mod native { log::error!( target: "nemo_relay.runtime", event = "event_sanitizer_panicked"; - "Event sanitizer panicked; preserving the last valid event snapshot" + "Event sanitizer panicked; clearing observability fields" ); - Some(fallback) + let mut cleared = fallback; + cleared.apply_sanitize_fields(EventSanitizeFields::default()); + Some(cleared) } }; (event, nested_publications) diff --git a/crates/core/src/api/shared.rs b/crates/core/src/api/shared.rs index d6eaf4b91..5d0652c8d 100644 --- a/crates/core/src/api/shared.rs +++ b/crates/core/src/api/shared.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use uuid::Uuid; -use crate::api::event::{Event, ScopeCategory}; +use crate::api::event::{Event, EventSanitizeFields, ScopeCategory}; use crate::api::llm::LlmRequest; use crate::api::registry::Guardrail; use crate::api::runtime::global_context; @@ -61,9 +61,13 @@ pub(crate) fn snapshot_event_sanitizers( log::error!( target: "nemo_relay.runtime", event = "event_sanitizer_snapshot_failed"; - "Event sanitizer snapshot failed open because the scope stack lock is poisoned; publishing without event sanitizers: {error}" + "Event sanitizer snapshot failed; clearing observability fields: {error}" ); - return None; + return Some(vec![Guardrail::new( + "event-sanitizer-snapshot-failure", + i32::MIN, + Arc::new(|_, _| Box::pin(async { Ok(EventSanitizeFields::default()) })), + )]); } }; let context = global_context(); @@ -73,9 +77,13 @@ pub(crate) fn snapshot_event_sanitizers( log::error!( target: "nemo_relay.runtime", event = "event_sanitizer_snapshot_failed"; - "Event sanitizer snapshot failed open because the runtime context lock is poisoned; publishing without event sanitizers: {error}" + "Event sanitizer snapshot failed; clearing observability fields: {error}" ); - return None; + return Some(vec![Guardrail::new( + "event-sanitizer-snapshot-failure", + i32::MIN, + Arc::new(|_, _| Box::pin(async { Ok(EventSanitizeFields::default()) })), + )]); } }; match &event { diff --git a/crates/core/src/api/tool.rs b/crates/core/src/api/tool.rs index 0d962549e..b2e55b141 100644 --- a/crates/core/src/api/tool.rs +++ b/crates/core/src/api/tool.rs @@ -245,7 +245,8 @@ pub struct ToolCallEndParams<'a> { /// /// # Notes /// Sanitize-request guardrails affect only the emitted start-event payload, not -/// the caller-owned `args` value. +/// the caller-owned `args` value. If a sanitizer errors or panics, Relay omits +/// that observability payload and does not run remaining sanitizers. pub fn tool_call(params: ToolCallParams<'_>) -> Result { ensure_runtime_owner()?; let scope_stack = current_scope_stack(); @@ -318,7 +319,7 @@ pub fn tool_call(params: ToolCallParams<'_>) -> Result { ) .await; let mut fields = event.sanitize_fields(); - fields.data = Some(sanitized); + fields.data = sanitized; event.apply_sanitize_fields(fields); event }) @@ -376,7 +377,7 @@ async fn tool_call_with_subscriber_snapshot( .timestamp_opt(params.timestamp) .build(); let handle = state.create_tool_handle(handle_params); - let event = state.build_tool_start_event(&handle, Some(sanitized_args)); + let event = state.build_tool_start_event(&handle, sanitized_args); let marks = skill_loads .into_iter() .map(|skill_load| { @@ -432,7 +433,8 @@ async fn tool_call_with_subscriber_snapshot( /// /// # Notes /// Sanitize-response guardrails affect only the emitted end-event payload, not -/// the caller-owned `result` value. +/// the caller-owned `result` value. If a sanitizer errors or panics, Relay omits +/// that observability payload and does not run remaining sanitizers. pub fn tool_call_end(params: ToolCallEndParams<'_>) -> Result<()> { ensure_runtime_owner()?; let scope_stack = current_scope_stack(); @@ -481,11 +483,13 @@ pub fn tool_call_end(params: ToolCallEndParams<'_>) -> Result<()> { ) .await; let mut fields = event.sanitize_fields(); - fields.data = if sanitized.is_null() { - fallback - } else { - Some(sanitized) - }; + fields.data = sanitized.and_then(|value| { + if value.is_null() { + fallback + } else { + Some(value) + } + }); event.apply_sanitize_fields(fields); event }) @@ -528,11 +532,13 @@ async fn tool_call_end_with_pending_marks( &entries, ) .await; - let data = if sanitized_result.is_null() { - params.data - } else { - Some(sanitized_result) - }; + let data = sanitized_result.and_then(|value| { + if value.is_null() { + params.data + } else { + Some(value) + } + }); let event = { let context = global_context(); let state = context diff --git a/crates/core/src/stream.rs b/crates/core/src/stream.rs index c48533afc..f6d3ca7b9 100644 --- a/crates/core/src/stream.rs +++ b/crates/core/src/stream.rs @@ -237,19 +237,19 @@ impl LlmStreamWrapper { aggregated }; - let entries = match self.scope_stack.read() { + let (entries, sanitizer_snapshot_failed) = match self.scope_stack.read() { Ok(scope_guard) => { let scope_locals = scope_guard .collect_scope_local_registries(|r| &r.llm_sanitize_response_guardrails); match global_context().read() { - Ok(state) => state.llm_sanitize_response_entries(&scope_locals), + Ok(state) => (state.llm_sanitize_response_entries(&scope_locals), false), Err(error) => { log::error!( target: "nemo_relay.runtime", event = "stream_end_sanitizer_snapshot_failed"; - "LLM stream END sanitizer snapshot failed open: {error}" + "LLM stream END sanitizer snapshot failed; omitting the observability payload: {error}" ); - Vec::new() + (Vec::new(), true) } } } @@ -257,9 +257,9 @@ impl LlmStreamWrapper { log::error!( target: "nemo_relay.runtime", event = "stream_end_sanitizer_snapshot_failed"; - "LLM stream END sanitizer snapshot failed open: {error}" + "LLM stream END sanitizer snapshot failed; omitting the observability payload: {error}" ); - Vec::new() + (Vec::new(), true) } }; let handle = self.handle.clone(); @@ -269,12 +269,17 @@ impl LlmStreamWrapper { let response_codec = self.response_codec.clone(); let sanitize_context = self.sanitize_context.clone(); let finalize = async move { - let sanitized = NemoRelayContextState::llm_sanitize_response_snapshot_chain( - response, - sanitize_context, - &entries, - ) - .await; + let sanitized = (!sanitizer_snapshot_failed).then(|| { + NemoRelayContextState::llm_sanitize_response_snapshot_chain( + response, + sanitize_context, + &entries, + ) + }); + let sanitized = match sanitized { + Some(sanitized) => sanitized.await, + None => None, + }; let data = match sanitized { Some(response) if response_was_null_without_fallback && response.is_null() => None, response => response, diff --git a/crates/core/tests/integration/native_plugin_tests.rs b/crates/core/tests/integration/native_plugin_tests.rs index 82fc2168e..8d572f247 100644 --- a/crates/core/tests/integration/native_plugin_tests.rs +++ b/crates/core/tests/integration/native_plugin_tests.rs @@ -932,7 +932,7 @@ async fn native_tool_execution_rejects_null_malformed_and_error_outcomes() { } #[tokio::test] -async fn native_event_sanitizer_callback_errors_preserve_observability_fields() { +async fn native_event_sanitizer_callback_errors_clear_observability_fields() { let _guard = NATIVE_PLUGIN_TEST_LOCK.lock().await; let fixture = build_fixture_plugin(); let manifest_ref = @@ -973,8 +973,8 @@ async fn native_event_sanitizer_callback_errors_preserve_observability_fields() let captured_events = events.lock().unwrap().clone(); let event = find_event(&captured_events, "native-event-sanitize-error", None); - assert_eq!(event.data(), Some(&json!({ "secret": true }))); - assert_eq!(event.metadata(), Some(&json!({ "secret": true }))); + assert_eq!(event.data(), None); + assert_eq!(event.metadata(), None); deregister_subscriber("native_event_sanitizer_error_capture") .expect("test subscriber should deregister"); diff --git a/crates/core/tests/integration/subscriber_dispatcher_tests.rs b/crates/core/tests/integration/subscriber_dispatcher_tests.rs index 1579d2cc1..7e4e6366d 100644 --- a/crates/core/tests/integration/subscriber_dispatcher_tests.rs +++ b/crates/core/tests/integration/subscriber_dispatcher_tests.rs @@ -368,7 +368,7 @@ fn dispatcher_continues_after_subscriber_panic() { } #[test] -fn dispatcher_publishes_the_snapshot_when_an_async_sanitizer_fails() { +fn dispatcher_clears_observability_fields_when_an_async_sanitizer_fails() { let _lock = TEST_MUTEX.lock().unwrap(); flush_subscribers().unwrap(); reset_global(); @@ -377,12 +377,12 @@ fn dispatcher_publishes_the_snapshot_when_an_async_sanitizer_fails() { let observed = Arc::new(Mutex::new(Vec::new())); let observed_events = Arc::clone(&observed); register_subscriber( - "fail-open-sanitizer-subscriber", + "fail-closed-sanitizer-subscriber", Arc::new(move |event| observed_events.lock().unwrap().push(event.clone())), ) .unwrap(); register_mark_sanitize_guardrail( - "fail-open-mark-sanitizer", + "fail-closed-mark-sanitizer", 10, Arc::new(|_, _| { Box::pin(async { @@ -407,17 +407,11 @@ fn dispatcher_publishes_the_snapshot_when_an_async_sanitizer_fails() { let observed = observed.lock().unwrap(); assert_eq!(observed.len(), 1); assert_eq!(observed[0].name(), "unsanitized-fallback"); - assert_eq!( - observed[0].sanitize_fields().data, - Some(json!({"original_data": true})) - ); - assert_eq!( - observed[0].sanitize_fields().metadata, - Some(json!({"original_metadata": true})) - ); + assert_eq!(observed[0].sanitize_fields().data, None); + assert_eq!(observed[0].sanitize_fields().metadata, None); drop(observed); - deregister_mark_sanitize_guardrail("fail-open-mark-sanitizer").unwrap(); - deregister_subscriber("fail-open-sanitizer-subscriber").unwrap(); + deregister_mark_sanitize_guardrail("fail-closed-mark-sanitizer").unwrap(); + deregister_subscriber("fail-closed-sanitizer-subscriber").unwrap(); } #[test] @@ -447,7 +441,7 @@ fn mark_emission_skips_sanitizers_without_subscribers() { } #[test] -fn dispatcher_publishes_the_snapshot_when_an_async_sanitizer_panics() { +fn dispatcher_clears_observability_fields_when_an_async_sanitizer_panics() { let _lock = TEST_MUTEX.lock().unwrap(); flush_subscribers().unwrap(); reset_global(); @@ -484,10 +478,7 @@ fn dispatcher_publishes_the_snapshot_when_an_async_sanitizer_panics() { let events = observed.lock().unwrap(); assert_eq!(events.len(), 1); assert_eq!(events[0].name(), "panic-fallback"); - assert_eq!( - events[0].sanitize_fields().data, - Some(json!({"redacted": true})) - ); + assert_eq!(events[0].sanitize_fields().data, None); drop(events); deregister_mark_sanitize_guardrail("successful-mark-sanitizer").unwrap(); deregister_mark_sanitize_guardrail("panic-mark-sanitizer").unwrap(); diff --git a/crates/core/tests/unit/dynamic_worker_tests.rs b/crates/core/tests/unit/dynamic_worker_tests.rs index 05bf45973..e4812eab5 100644 --- a/crates/core/tests/unit/dynamic_worker_tests.rs +++ b/crates/core/tests/unit/dynamic_worker_tests.rs @@ -1574,8 +1574,8 @@ async fn installed_callbacks_apply_surface_specific_fallbacks() { for entries in [mark_entries, scope_start_entries, scope_end_entries] { let sanitized = NemoRelayContextState::event_sanitize_snapshot_chain(event.clone(), &entries).await; - assert_eq!(sanitized.data(), event.data()); - assert_eq!(sanitized.metadata(), event.metadata()); + assert_eq!(sanitized.data(), None); + assert_eq!(sanitized.metadata(), None); } assert_eq!( @@ -1585,7 +1585,7 @@ async fn installed_callbacks_apply_surface_specific_fallbacks() { &tool_request_entries, ) .await, - tool_request + None ); assert_eq!( NemoRelayContextState::tool_sanitize_response_snapshot_chain( @@ -1594,7 +1594,7 @@ async fn installed_callbacks_apply_surface_specific_fallbacks() { &tool_response_entries, ) .await, - tool_response + None ); assert_eq!( NemoRelayContextState::llm_sanitize_request_snapshot_chain( @@ -1603,7 +1603,7 @@ async fn installed_callbacks_apply_surface_specific_fallbacks() { &llm_request_entries, ) .await, - Some(llm_request), + None, ); assert_eq!( NemoRelayContextState::llm_sanitize_response_snapshot_chain( @@ -1612,7 +1612,7 @@ async fn installed_callbacks_apply_surface_specific_fallbacks() { &llm_response_entries, ) .await, - Some(llm_response), + None, ); crate::api::subscriber::flush_subscribers().expect("subscriber callback should flush"); } diff --git a/crates/core/tests/unit/runtime_state_tests.rs b/crates/core/tests/unit/runtime_state_tests.rs index 09f01f279..9fb8cb2d5 100644 --- a/crates/core/tests/unit/runtime_state_tests.rs +++ b/crates/core/tests/unit/runtime_state_tests.rs @@ -3,13 +3,16 @@ //! Unit tests for runtime middleware snapshot chains. +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + use serde_json::{Map, json}; use super::*; use crate::api::registry::{RegistryRecord, RequestIntercept}; #[tokio::test] -async fn middleware_snapshot_chains_contain_callback_panics() { +async fn sanitizer_snapshot_chains_fail_closed_on_callback_panics() { let event = Event::Mark(MarkEvent::new( BaseEvent::builder() .name("preserved-event") @@ -21,18 +24,37 @@ async fn middleware_snapshot_chains_contain_callback_panics() { )); let event_sanitizer: EventSanitizeFn = Arc::new(|_, _| Box::pin(async { panic!("event sanitizer panic") })); + let event_later_called = Arc::new(AtomicBool::new(false)); + let event_later_called_by_callback = Arc::clone(&event_later_called); + let event_later_sanitizer: EventSanitizeFn = Arc::new(move |_, fields| { + event_later_called_by_callback.store(true, Ordering::Release); + Box::pin(async move { Ok(fields) }) + }); let sanitized_event = NemoRelayContextState::event_sanitize_snapshot_chain( event.clone(), - &[RegistryRecord::new("event-panic", 0, event_sanitizer)], + &[ + RegistryRecord::new("event-panic", 0, event_sanitizer), + RegistryRecord::new("event-later", -1, event_later_sanitizer), + ], ) .await; - assert_eq!(sanitized_event.data(), event.data()); - assert_eq!(sanitized_event.metadata(), event.metadata()); + assert_eq!(sanitized_event.data(), None); + assert_eq!(sanitized_event.metadata(), None); + assert!(!event_later_called.load(Ordering::Acquire)); let tool_payload = json!({"tool": "preserved"}); let tool_sanitizer: ToolSanitizeFn = Arc::new(|_, _| Box::pin(async { panic!("tool sanitizer panic") })); - let tool_entries = vec![RegistryRecord::new("tool-panic", 0, tool_sanitizer)]; + let tool_later_called = Arc::new(AtomicBool::new(false)); + let tool_later_called_by_callback = Arc::clone(&tool_later_called); + let tool_later_sanitizer: ToolSanitizeFn = Arc::new(move |_, value| { + tool_later_called_by_callback.store(true, Ordering::Release); + Box::pin(async move { Ok(value) }) + }); + let tool_entries = vec![ + RegistryRecord::new("tool-panic", 0, tool_sanitizer), + RegistryRecord::new("tool-later", -1, tool_later_sanitizer), + ]; assert_eq!( NemoRelayContextState::tool_sanitize_request_snapshot_chain( "tool", @@ -40,24 +62,31 @@ async fn middleware_snapshot_chains_contain_callback_panics() { &tool_entries, ) .await, - tool_payload + None ); + assert!(!tool_later_called.load(Ordering::Acquire)); let tool_response = json!({"tool_response": "preserved"}); let tool_response_sanitizer: ToolSanitizeFn = Arc::new(|_, _| Box::pin(async { panic!("tool response sanitizer panic") })); + let tool_response_later_called = Arc::new(AtomicBool::new(false)); + let tool_response_later_called_by_callback = Arc::clone(&tool_response_later_called); + let tool_response_later_sanitizer: ToolSanitizeFn = Arc::new(move |_, value| { + tool_response_later_called_by_callback.store(true, Ordering::Release); + Box::pin(async move { Ok(value) }) + }); assert_eq!( NemoRelayContextState::tool_sanitize_response_snapshot_chain( "tool", tool_response.clone(), - &[RegistryRecord::new( - "tool-response-panic", - 0, - tool_response_sanitizer, - )], + &[ + RegistryRecord::new("tool-response-panic", 0, tool_response_sanitizer,), + RegistryRecord::new("tool-response-later", -1, tool_response_later_sanitizer,) + ], ) .await, - tool_response + None ); + assert!(!tool_response_later_called.load(Ordering::Acquire)); let request = LlmRequest { headers: Map::new(), @@ -65,7 +94,16 @@ async fn middleware_snapshot_chains_contain_callback_panics() { }; let llm_sanitizer: LlmSanitizeRequestFn = Arc::new(|_, _| Box::pin(async { panic!("LLM sanitizer panic") })); - let llm_entries = vec![RegistryRecord::new("llm-panic", 0, llm_sanitizer)]; + let llm_later_called = Arc::new(AtomicBool::new(false)); + let llm_later_called_by_callback = Arc::clone(&llm_later_called); + let llm_later_sanitizer: LlmSanitizeRequestFn = Arc::new(move |request, _| { + llm_later_called_by_callback.store(true, Ordering::Release); + Box::pin(async move { Ok(Some(request)) }) + }); + let llm_entries = vec![ + RegistryRecord::new("llm-panic", 0, llm_sanitizer), + RegistryRecord::new("llm-later", -1, llm_later_sanitizer), + ]; assert_eq!( NemoRelayContextState::llm_sanitize_request_snapshot_chain( request.clone(), @@ -73,24 +111,31 @@ async fn middleware_snapshot_chains_contain_callback_panics() { &llm_entries, ) .await, - Some(request.clone()) + None ); + assert!(!llm_later_called.load(Ordering::Acquire)); let llm_response = json!({"llm_response": "preserved"}); let llm_response_sanitizer: LlmSanitizeResponseFn = Arc::new(|_, _| Box::pin(async { panic!("LLM response sanitizer panic") })); + let llm_response_later_called = Arc::new(AtomicBool::new(false)); + let llm_response_later_called_by_callback = Arc::clone(&llm_response_later_called); + let llm_response_later_sanitizer: LlmSanitizeResponseFn = Arc::new(move |response, _| { + llm_response_later_called_by_callback.store(true, Ordering::Release); + Box::pin(async move { Ok(Some(response)) }) + }); assert_eq!( NemoRelayContextState::llm_sanitize_response_snapshot_chain( llm_response.clone(), LlmSanitizeResponseContext::default(), - &[RegistryRecord::new( - "llm-response-panic", - 0, - llm_response_sanitizer, - )], + &[ + RegistryRecord::new("llm-response-panic", 0, llm_response_sanitizer,), + RegistryRecord::new("llm-response-later", -1, llm_response_later_sanitizer,) + ], ) .await, - Some(llm_response) + None ); + assert!(!llm_response_later_called.load(Ordering::Acquire)); let tool_conditional: ToolConditionalFn = Arc::new(|_, _| Box::pin(async { panic!("tool conditional panic") })); diff --git a/crates/core/tests/unit/subscriber_dispatcher_tests.rs b/crates/core/tests/unit/subscriber_dispatcher_tests.rs index dc242cf77..3a8f2c883 100644 --- a/crates/core/tests/unit/subscriber_dispatcher_tests.rs +++ b/crates/core/tests/unit/subscriber_dispatcher_tests.rs @@ -532,7 +532,7 @@ fn detached_publications_share_one_background_executor_thread() { } #[test] -fn sanitizer_runtime_failure_preserves_untransformed_event_snapshot() { +fn sanitizer_runtime_failure_clears_untransformed_event_fields() { let _lock = crate::shared_runtime::runtime_owner_test_mutex() .lock() .unwrap_or_else(|error| error.into_inner()); @@ -541,7 +541,9 @@ fn sanitizer_runtime_failure_preserves_untransformed_event_snapshot() { "atof_version": "0.1", "uuid": "019c1df6-4a57-7000-8000-000000000008", "timestamp": "2026-07-28T00:00:00Z", - "name": "fail-open-runtime" + "name": "fail-closed-runtime", + "data": {"secret": true}, + "metadata": {"secret": true} })) .expect("valid event"); let sanitizer: EventSanitizeFn = Arc::new(|_, _| { @@ -559,7 +561,10 @@ fn sanitizer_runtime_failure_preserves_untransformed_event_snapshot() { ); set_sanitizer_runtime_failure_for_test(None); - assert_eq!(published, Some(event)); + let published = published.expect("sanitizer runtime failure still publishes the event shell"); + assert_eq!(published.name(), event.name()); + assert_eq!(published.data(), None); + assert_eq!(published.metadata(), None); assert!(nested.is_empty()); } diff --git a/crates/ffi/tests/unit/api/registry_tests.rs b/crates/ffi/tests/unit/api/registry_tests.rs index 57caea2e2..ae1bc656b 100644 --- a/crates/ffi/tests/unit/api/registry_tests.rs +++ b/crates/ffi/tests/unit/api/registry_tests.rs @@ -436,8 +436,12 @@ fn test_ffi_event_sanitizer_registries_and_error_paths() { .iter() .find(|event| event["name"] == "ffi-invalid-callback-mark") .expect("invalid callback mark should be delivered"); - assert_eq!(invalid_callback_event["data"], json!({"secret": true})); - assert_eq!(invalid_callback_event["metadata"], json!({"secret": true})); + assert_eq!(invalid_callback_event["data"], Json::Null); + assert_eq!( + invalid_callback_event["json"]["category_profile"], + Json::Null + ); + assert_eq!(invalid_callback_event["metadata"], Json::Null); for name in ["ffi-local-child", "ffi-local-mark"] { for event in events.iter().filter(|event| event["name"] == name) { assert_eq!(event["data"], json!({"sanitized_by": name})); diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index ecd41f2da..bf1f468ac 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -2822,8 +2822,9 @@ macro_rules! napi_event_guardrail_api { /// The callback may return fields directly or in a Promise. Scope and mark /// calls queue the event and return synchronously; publication resumes after /// the Promise settles. Callback, serialization, conversion, or invalid-result - /// failures preserve the last valid event fields and record the error for - /// `getLastCallbackError()`. + /// failures clear the emitted event fields and record the error for + /// `getLastCallbackError()`. Await `flushSubscribers()` before inspecting + /// either the delivered event or that error. #[napi] pub fn $register_name( env: Env, @@ -2901,7 +2902,7 @@ 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 + /// If the callback throws, Relay omits the emitted payload and records the error /// for `getLastCallbackError()`. register_tool_sanitize_request_guardrail, /// Deregister a tool request sanitization guardrail by name. @@ -2918,7 +2919,7 @@ 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 + /// If the callback throws, Relay omits the emitted payload and records the error /// for `getLastCallbackError()`. register_tool_sanitize_response_guardrail, /// Deregister a tool response sanitization guardrail by name. @@ -3071,8 +3072,8 @@ pub fn deregister_tool_execution_intercept(name: String) -> Result { /// /// The `guardrail` callback receives `(request, context)` and must return the sanitized request, /// or `null` to omit the observability payload. Lower `priority` values run first. Throws if a -/// guardrail with the same `name` already exists. If the callback throws, Relay preserves the last -/// valid payload, continues publication, and records the error for `getLastCallbackError()`. +/// guardrail with the same `name` already exists. If the callback throws, Relay omits the payload +/// and annotation, continues publication, and records the error for `getLastCallbackError()`. #[napi] pub fn register_llm_sanitize_request_guardrail( env: Env, @@ -3105,8 +3106,8 @@ pub fn deregister_llm_sanitize_request_guardrail(name: String) -> Result { /// /// The `guardrail` callback receives `(response, context)` and must return the sanitized response, /// or `null` to omit the observability payload. Lower `priority` values run first. Throws if a -/// guardrail with the same `name` already exists. If the callback throws, Relay preserves the last -/// valid payload, continues publication, and records the error for `getLastCallbackError()`. +/// guardrail with the same `name` already exists. If the callback throws, Relay omits the payload +/// and annotation, continues publication, and records the error for `getLastCallbackError()`. #[napi] pub fn register_llm_sanitize_response_guardrail( env: Env, @@ -3364,8 +3365,9 @@ macro_rules! napi_scope_event_guardrail_api { /// The callback may return fields directly or in a Promise. Scope and mark /// calls queue the event and return synchronously; publication resumes after /// the Promise settles. Callback, serialization, conversion, or invalid-result - /// failures preserve the last valid event fields and record the error for - /// `getLastCallbackError()`. + /// failures clear the emitted event fields and record the error for + /// `getLastCallbackError()`. Await `flushSubscribers()` before inspecting + /// either the delivered event or that error. #[napi] pub fn $register_name( env: Env, @@ -3460,7 +3462,7 @@ 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 + /// If the callback throws, Relay omits the emitted payload and records the error /// for `getLastCallbackError()`. scope_register_tool_sanitize_request_guardrail, /// Deregister a scope-local tool request sanitization guardrail by name. @@ -3478,7 +3480,7 @@ 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 + /// If the callback throws, Relay omits the emitted payload and records the error /// for `getLastCallbackError()`. scope_register_tool_sanitize_response_guardrail, /// Deregister a scope-local tool response sanitization guardrail by name. @@ -3653,7 +3655,7 @@ pub fn scope_deregister_tool_execution_intercept(scope_uuid: String, name: Strin /// The `guardrail` callback receives `(request, context)` and must return the sanitized request, /// or `null` to omit the observability payload. Lower `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 last valid payload, continues publication, and records the error for +/// Relay omits the payload and annotation, continues publication, and records the error for /// `getLastCallbackError()`. #[napi] pub fn scope_register_llm_sanitize_request_guardrail( @@ -3698,7 +3700,7 @@ pub fn scope_deregister_llm_sanitize_request_guardrail( /// The `guardrail` callback receives `(response, context)` and must return the sanitized response, /// or `null` to omit the observability payload. Lower `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 last valid payload, continues publication, and records the error for +/// Relay omits the payload and annotation, continues publication, and records the error for /// `getLastCallbackError()`. #[napi] pub fn scope_register_llm_sanitize_response_guardrail( diff --git a/crates/node/src/callable.rs b/crates/node/src/callable.rs index f085fb0fa..28e6384a2 100644 --- a/crates/node/src/callable.rs +++ b/crates/node/src/callable.rs @@ -572,7 +572,7 @@ pub fn wrap_js_event_sanitize_promise_fn(func: Arc) -> EventSani } .inspect_err(|error| { // Scope and mark publication happens on the dispatcher - // thread. Preserve the event (the core fails open) while + // thread. The core clears the governed observability fields while // making the binding-visible failure available to Node. record_callback_error(error.to_string()); })?; diff --git a/crates/node/tests/event_sanitizers_tests.mjs b/crates/node/tests/event_sanitizers_tests.mjs index c1d62acf7..ec6bc3327 100644 --- a/crates/node/tests/event_sanitizers_tests.mjs +++ b/crates/node/tests/event_sanitizers_tests.mjs @@ -31,6 +31,12 @@ function assertSanitizerFieldsPreserved(event, expectedData, expectedMetadata = assert.deepEqual(event.metadata, expectedMetadata); } +function assertSanitizerFieldsCleared(event) { + assert.equal(event.data, null); + assert.equal(event.category_profile, null); + assert.equal(event.metadata, null); +} + async function initializeWithoutDiscoveredPluginConfig(config) { const previousDirectory = process.cwd(); const directory = mkdtempSync(path.join(tmpdir(), 'nemo-relay-node-')); @@ -361,7 +367,7 @@ describe('event sanitizer registries', () => { } }); - it('fails open and records invalid sanitizer results', async () => { + it('fails closed and records invalid sanitizer results', async () => { const events = capture('node-event-sanitize-invalid-sub'); const invalidResults = { scalar: () => 'invalid', @@ -389,7 +395,7 @@ describe('event sanitizer registries', () => { lib.deregisterMarkSanitizeGuardrail(seedName); lib.deregisterMarkSanitizeGuardrail(name); } - assertSanitizerFieldsPreserved(events.at(-1), { kept: kind }); + assertSanitizerFieldsCleared(events.at(-1)); assert.match(lib.getLastCallbackError(), /invalid JS event sanitizer result/); } } finally { @@ -417,7 +423,7 @@ describe('event sanitizer registries', () => { assert.equal(start.metadata.background, true); }); - it('fails open and records invalid queued sanitizer results', async () => { + it('fails closed and records invalid queued sanitizer results', async () => { const events = capture('node-event-sanitize-background-invalid-sub'); const invalidResults = { emptyObject: () => ({}), @@ -447,7 +453,7 @@ describe('event sanitizer registries', () => { const start = events.find( (event) => event.kind === 'scope' && event.name === name && event.scope_category === 'start', ); - assertSanitizerFieldsPreserved(start, { kept: kind }); + assertSanitizerFieldsCleared(start); assert.match(lib.getLastCallbackError(), /invalid JS event sanitizer result/); } } finally { @@ -455,7 +461,7 @@ describe('event sanitizer registries', () => { } }); - it('fails open when a queued sanitizer throws', async () => { + it('fails closed when a queued sanitizer throws', async () => { const events = capture('node-event-sanitize-background-throw-sub'); lib.clearLastCallbackError(); lib.registerScopeSanitizeStartGuardrail('node-background-throw-seed', -1, (_event, fields) => ({ @@ -474,7 +480,7 @@ describe('event sanitizer registries', () => { const start = events.find( (event) => event.kind === 'scope' && event.name === 'background-throw-tool' && event.scope_category === 'start', ); - assertSanitizerFieldsPreserved(start, { kept: true }); + assertSanitizerFieldsCleared(start); assert.match(lib.getLastCallbackError() ?? '', /background sanitizer boom/i); } finally { lib.deregisterScopeSanitizeStartGuardrail('node-background-throw-seed'); @@ -543,7 +549,7 @@ describe('event sanitizer registries', () => { assert.deepEqual(marks.cleared.data, { raw: true }); }); - it('fails open when a plugin-owned sanitizer throws', async () => { + it('fails closed 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, { @@ -568,7 +574,7 @@ describe('event sanitizer registries', () => { lib.event('plugin-throw', null, { raw: true }, { raw: true }); await lib.flushSubscribers(); await waitFor(events, 1); - assertSanitizerFieldsPreserved(events.at(-1), { raw: true }); + assertSanitizerFieldsCleared(events.at(-1)); assert.match(lib.getLastCallbackError() ?? '', /plugin sanitizer boom/i); } finally { plugin.clear(); diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index 250aa0c7d..fbba803f7 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/crates/node/tests/llm_tests.mjs @@ -831,7 +831,7 @@ describe('LLM guardrails', () => { } }); - it('sanitize request guardrail failures preserve the payload and remain usable', async () => { + it('sanitize request guardrail failures omit the payload and remain usable', async () => { const events = []; clearLastCallbackError(); registerSubscriber('node_llm_san_req_throw_sub', (event) => events.push(event)); @@ -849,7 +849,7 @@ describe('LLM guardrails', () => { event.category === 'llm' && event.scope_category === 'start', ); - assert.deepEqual(start.data, { headers: request.headers, content: request.content }); + assert.equal(start.data, null); assert.equal(getLastCallbackError(), 'internal error: unknown error'); deregisterLlmSanitizeRequestGuardrail('node_llm_san_req_throw'); @@ -944,7 +944,7 @@ describe('LLM guardrails', () => { } }); - it('sanitize response guardrail failures preserve the payload and remain usable', async () => { + it('sanitize response guardrail failures omit the payload and remain usable', async () => { const events = []; clearLastCallbackError(); registerSubscriber('node_llm_san_resp_throw_sub', (event) => events.push(event)); @@ -962,7 +962,7 @@ describe('LLM guardrails', () => { event.category === 'llm' && event.scope_category === 'end', ); - assert.deepEqual(end.data, response); + assert.equal(end.data, null); assert.match(getLastCallbackError() ?? '', /response sanitizer boom/i); deregisterLlmSanitizeResponseGuardrail('node_llm_san_resp_throw'); diff --git a/crates/node/tests/tools_tests.mjs b/crates/node/tests/tools_tests.mjs index d40a7f6e2..c7ce4443d 100644 --- a/crates/node/tests/tools_tests.mjs +++ b/crates/node/tests/tools_tests.mjs @@ -680,7 +680,7 @@ describe('Tool guardrails', () => { } }); - it('sanitize guardrail failures preserve the observable payload', async () => { + it('sanitize guardrail failures omit the observable payload', async () => { clearLastCallbackError(); const events = []; registerSubscriber('node_tool_san_throw_sub', (event) => events.push(event)); @@ -697,7 +697,7 @@ describe('Tool guardrails', () => { event.category === 'tool' && event.scope_category === 'start', ); - assert.deepEqual(start.data, { original: true }); + assert.equal(start.data, null); assert.match(getLastCallbackError() ?? '', /sanitize boom/i); } finally { deregisterToolSanitizeRequestGuardrail('node_tool_san_throw'); diff --git a/crates/python/src/py_callable.rs b/crates/python/src/py_callable.rs index 85d468d2b..db7b727cb 100644 --- a/crates/python/src/py_callable.rs +++ b/crates/python/src/py_callable.rs @@ -837,11 +837,17 @@ pub fn wrap_py_tool_fn(py_fn: Py) -> ToolSanitizeFn { invocation_context.as_ref(), ) })) - .await?; - Python::attach(|py| { - py_to_json(result.bind(py)) - .map_err(|e| FlowError::Internal(format!("tool py_to_json failed: {e}"))) - }) + .await + .and_then(|result| { + Python::attach(|py| { + py_to_json(result.bind(py)) + .map_err(|e| FlowError::Internal(format!("tool py_to_json failed: {e}"))) + }) + }); + if let Err(error) = &result { + eprintln!("nemo_relay: Python tool sanitizer callable failed: {error}"); + } + result }) }) } @@ -1339,21 +1345,27 @@ fn wrap_py_llm_sanitize_request_callback(py_fn: Py) -> LlmSanitizeRequest invocation_context.as_ref(), ) })) - .await?; - Python::attach(|py| { - if result.is_none(py) { - Ok(None) - } else { - result - .extract::(py) - .map(|request| Some(request.inner)) - .map_err(|error| { - FlowError::Internal(format!( - "LLM sanitize request returned unexpected type: {error}" - )) - }) - } - }) + .await + .and_then(|result| { + Python::attach(|py| { + if result.is_none(py) { + Ok(None) + } else { + result + .extract::(py) + .map(|request| Some(request.inner)) + .map_err(|error| { + FlowError::Internal(format!( + "LLM sanitize request returned unexpected type: {error}" + )) + }) + } + }) + }); + if let Err(error) = &result { + eprintln!("nemo_relay: Python LLM sanitize request callable failed: {error}"); + } + result }) }, ) @@ -1658,16 +1670,22 @@ fn wrap_py_llm_sanitize_response_callback(py_fn: Py) -> LlmSanitizeRespon invocation_context.as_ref(), ) })) - .await?; - Python::attach(|py| { - if result.is_none(py) { - Ok(None) - } else { - py_to_json(result.bind(py)) - .map(Some) - .map_err(|error| FlowError::Internal(error.to_string())) - } - }) + .await + .and_then(|result| { + Python::attach(|py| { + if result.is_none(py) { + Ok(None) + } else { + py_to_json(result.bind(py)) + .map(Some) + .map_err(|error| FlowError::Internal(error.to_string())) + } + }) + }); + if let Err(error) = &result { + eprintln!("nemo_relay: Python LLM sanitize response callable failed: {error}"); + } + result }) }) } @@ -1796,16 +1814,25 @@ pub fn wrap_py_event_sanitize_fn(py_fn: Py) -> EventSanitizeFn { ) }, ); - let result = resolve_py_object_or_future(result).await?; - Python::attach(|py| { - py_to_json(result.bind(py)) - .map_err(|error| FlowError::Internal(error.to_string())) - .and_then(|value| { - serde_json::from_value(value).map_err(|error| { - FlowError::Internal(format!("invalid event sanitizer result: {error}")) - }) + let result = resolve_py_object_or_future(result) + .await + .and_then(|result| { + Python::attach(|py| { + py_to_json(result.bind(py)) + .map_err(|error| FlowError::Internal(error.to_string())) + .and_then(|value| { + serde_json::from_value(value).map_err(|error| { + FlowError::Internal(format!( + "invalid event sanitizer result: {error}" + )) + }) + }) }) - }) + }); + if let Err(error) = &result { + eprintln!("nemo_relay: Python event sanitizer callable failed: {error}"); + } + result }) }) } diff --git a/docs/about-nemo-relay/concepts/middleware.mdx b/docs/about-nemo-relay/concepts/middleware.mdx index 2a3aabe94..55caa6064 100644 --- a/docs/about-nemo-relay/concepts/middleware.mdx +++ b/docs/about-nemo-relay/concepts/middleware.mdx @@ -193,8 +193,8 @@ event later in FIFO order. Subscriber and exporter delivery is therefore delayed, while start/end/mark order is preserved. Closing a scope or deregistering middleware after emission -does not affect queued snapshots. Sanitizer failures fail open: Relay records -the callback failure and publishes the last valid event snapshot. +does not affect queued snapshots. Sanitizer failures fail closed: Relay records +the callback failure and withholds the governed observability payload. ## Managed Execution Order diff --git a/docs/about-nemo-relay/release-notes/index.mdx b/docs/about-nemo-relay/release-notes/index.mdx index d36bc2afc..0a9f80ba6 100644 --- a/docs/about-nemo-relay/release-notes/index.mdx +++ b/docs/about-nemo-relay/release-notes/index.mdx @@ -36,6 +36,9 @@ compatibility information applies to the current 0.7 prerelease. for each managed call. Sanitizers can normalize built-in, runtime-registered, and opaque codec payloads without changing the client-visible request or response. +- Sanitizer callback failures now fail closed across mark, scope, tool, and + LLM surfaces: Relay withholds the governed observability payload while + preserving application execution. - The PII redaction plugin now selects the active codec per call. One component can safely sanitize mixed OpenAI Chat, OpenAI Responses, and Anthropic Messages traffic without a fixed provider codec. diff --git a/docs/build-plugins/dynamic-plugins/native-dynamic/about.mdx b/docs/build-plugins/dynamic-plugins/native-dynamic/about.mdx index 18f2d0adc..82167c28e 100644 --- a/docs/build-plugins/dynamic-plugins/native-dynamic/about.mdx +++ b/docs/build-plugins/dynamic-plugins/native-dynamic/about.mdx @@ -243,9 +243,9 @@ JSON string handles, which callers release with the ordinary host string release operation. A successful null sanitizer output omits the LLM observability payload and its -annotation. Returning an error fails open: Relay records the callback error and -preserves the last valid observability payload and annotation. Neither case -changes the client-visible request or response. +annotation. Returning an error also omits the payload and annotation; Relay +records the callback error. Neither case changes the client-visible request or +response. Use the `nemo-relay-plugin` crate rather than the host `nemo-relay` runtime crate. Refer to [Build a Rust Native Plugin](/build-plugins/dynamic-plugins/native-dynamic/rust-native-plugin-example) diff --git a/docs/instrument-applications/advanced-guide.mdx b/docs/instrument-applications/advanced-guide.mdx index 70220e456..72737f7cf 100644 --- a/docs/instrument-applications/advanced-guide.mdx +++ b/docs/instrument-applications/advanced-guide.mdx @@ -202,8 +202,8 @@ Node.js guardrail and request-intercept callbacks can return direct values or Promises. A thrown error or rejected Promise from a conditional-execution guardrail or request intercept rejects the managed call before protected execution or later middleware runs. A thrown error or rejected Promise from a -sanitize guardrail or event sanitizer fails open, preserving the current emitted -payload and recording the error for `getLastCallbackError()`. +sanitize guardrail or event sanitizer fails closed, withholding the governed +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. diff --git a/docs/reference/event-sanitizers.mdx b/docs/reference/event-sanitizers.mdx index e4bb328a7..aea5e537b 100644 --- a/docs/reference/event-sanitizers.mdx +++ b/docs/reference/event-sanitizers.mdx @@ -52,12 +52,13 @@ side effects, such as metrics, logging, mutation, or I/O; Relay only guarantees the transformed observability payload. 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. The same -rule applies to tool and LLM request/response sanitizer errors: Relay preserves -the last valid observability payload without changing provider execution. In Node.js, -a synchronous sanitizer callback that throws also fails open; Relay records the -error for `getLastCallbackError()`. +callback receives the fields returned by the callback before it. A callback +error, panic, rejected Promise, or invalid binding result fails closed and +short-circuits the remaining sanitizer chain. Event sanitizers clear `data`, +`category_profile`, and `metadata`; tool and LLM sanitizers omit their governed +payload, while LLM sanitizers also omit its annotation. Sanitizer failures do +not change tool or provider execution. Node.js records the error for +`getLastCallbackError()`. Event sanitizer callbacks are not re-entrant. Do not call another NeMo Relay @@ -80,7 +81,7 @@ exporters. This preserves FIFO start/end/mark delivery without making `push_scope`, `pop_scope`, or `event` awaitable. A scope-local sanitizer removed after an event is emitted still applies to its queued snapshot. An asynchronous -sanitizer rejection fails open and preserves the last valid event fields. +sanitizer rejection fails closed and clears the governed observability fields. When Python code emits events from an `asyncio` task, await `nemo_relay.subscribers.flush_async()` to drain queued publication without diff --git a/docs/reference/migration-guides.mdx b/docs/reference/migration-guides.mdx index 91d95bdf2..e50a6762a 100644 --- a/docs/reference/migration-guides.mdx +++ b/docs/reference/migration-guides.mdx @@ -116,9 +116,9 @@ includes `push_scope`, `pop_scope`, mark APIs, `tool_call`, `tool_call_end`, sanitizer/subscriber chain, then enqueue sanitization and publication on a serial dispatcher. Event subscribers and exporters therefore receive sanitized events later, in emission order. Do not add `await` to these lifecycle calls. -Only enqueue-time validation and runtime-state errors are returned directly; -middleware and codec errors discovered during queued publication are logged and -handled according to their fail-open contracts. +Only enqueue-time validation and runtime-state errors are returned directly. +Sanitizer errors discovered during queued publication are logged and fail +closed; codec errors retain their documented fallback behavior. Python subscriber flushing is context-sensitive. Continue to call `nemo_relay.subscribers.flush()` from synchronous code. From a running @@ -219,9 +219,8 @@ Check every callback for an implicit empty return. In particular: - A Python function that reaches the end without `return` omits the payload. - A JavaScript callback that returns `null` or `undefined` omits the payload. - A Rust callback must return `Some(payload)` to retain the payload. -- A sanitizer error or panic fails open: Relay preserves the last valid payload - and annotation snapshot, logs or records the callback error, and continues - publication. +- A sanitizer error or panic fails closed: Relay omits the payload and + annotation, logs or records the callback error, and continues publication. Use omission only when recording the payload would be unsafe. diff --git a/go/nemo_relay/event_sanitizers_test.go b/go/nemo_relay/event_sanitizers_test.go index 0a0b91426..557a6bde9 100644 --- a/go/nemo_relay/event_sanitizers_test.go +++ b/go/nemo_relay/event_sanitizers_test.go @@ -25,12 +25,20 @@ func TestEventSanitizerRegistries(t *testing.T) { runTestWithScopeStack(t, testEventSanitizerRegistries) } -func TestEventSanitizerMarshalFailurePreservesObservabilityFields(t *testing.T) { +func TestEventSanitizerMarshalFailureClearsObservabilityFields(t *testing.T) { runTestWithScopeStack(t, func(t *testing.T) { var mu sync.Mutex var events []Event registerEventSanitizerSubscriber(t, &mu, &events) + if err := RegisterMarkSanitizeGuardrail("go-mark-sanitize-category-profile", 1, func(_ Event, fields EventSanitizeFields) EventSanitizeFields { + fields.CategoryProfile = json.RawMessage(`{"subtype":"seeded"}`) + return fields + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = DeregisterMarkSanitizeGuardrail("go-mark-sanitize-category-profile") }) + if err := RegisterMarkSanitizeGuardrail("go-mark-sanitize-invalid", 0, func(_ Event, _ EventSanitizeFields) EventSanitizeFields { return EventSanitizeFields{Data: json.RawMessage("{")} }); err != nil { @@ -50,10 +58,10 @@ func TestEventSanitizerMarshalFailurePreservesObservabilityFields(t *testing.T) if len(events) != 1 { t.Fatalf("expected one event, got %d", len(events)) } - if string(events[0].Data()) != `{"secret":true}` || + if len(events[0].Data()) != 0 || len(events[0].CategoryProfile()) != 0 || - string(events[0].Metadata()) != `{"secret":true}` { - t.Fatalf("expected the last valid observability fields, got data=%s category_profile=%s metadata=%s", events[0].Data(), events[0].CategoryProfile(), events[0].Metadata()) + len(events[0].Metadata()) != 0 { + t.Fatalf("expected cleared observability fields, got data=%s category_profile=%s metadata=%s", events[0].Data(), events[0].CategoryProfile(), events[0].Metadata()) } }) } diff --git a/python/nemo_relay/__init__.py b/python/nemo_relay/__init__.py index fe3b9f43f..585fbe683 100644 --- a/python/nemo_relay/__init__.py +++ b/python/nemo_relay/__init__.py @@ -166,9 +166,12 @@ class EventSanitizeFields(TypedDict): #: Guardrail callback that sanitizes emitted tool request or response payloads. #: Arguments are the tool name and JSON payload. The return value is the JSON -#: payload recorded on the emitted event. Exceptions fail open and preserve the -#: last valid observability payload. +#: payload recorded on the emitted event. Exceptions fail closed and omit the +#: observability payload, then stop the remaining sanitizer chain. ToolSanitizeGuardrail: TypeAlias = Callable[[str, Json], Json | Awaitable[Json]] +#: Guardrail callback that sanitizes mark and scope event fields. Exceptions +#: fail closed: Relay clears all mutable observability fields and stops the +#: remaining sanitizer chain. EventSanitizeGuardrail: TypeAlias = Callable[ ["Event", EventSanitizeFields], EventSanitizeFields | Awaitable[EventSanitizeFields] ] @@ -177,7 +180,8 @@ class EventSanitizeFields(TypedDict): ToolConditionalExecutionGuardrail: TypeAlias = Callable[[str, Json], Optional[str] | Awaitable[Optional[str]]] #: Guardrail callback that sanitizes an ``LLMRequest`` used for emitted events. #: Callbacks receive ``(request, context)``. Returning ``None`` omits the LLM observability -#: payload and annotation without changing the caller-visible request. +#: payload and annotation without changing the caller-visible request. Exceptions +#: also fail closed and stop the remaining sanitizer chain. LlmSanitizeRequestGuardrail: TypeAlias = Callable[ [LLMRequest, "LlmSanitizeRequestContext"], Optional[LLMRequest] | Awaitable[Optional[LLMRequest]], @@ -185,6 +189,7 @@ class EventSanitizeFields(TypedDict): #: Guardrail callback that sanitizes an emitted JSON LLM response payload. #: Callbacks receive ``(response, context)`` and can return ``None`` to omit #: observability payload and annotation without changing the caller response. +#: Exceptions also fail closed and stop the remaining sanitizer chain. LlmSanitizeResponseGuardrail: TypeAlias = Callable[ [Json, "LlmSanitizeResponseContext"], Optional[Json] | Awaitable[Optional[Json]] ] diff --git a/python/nemo_relay/__init__.pyi b/python/nemo_relay/__init__.pyi index 31dee7ea2..ed0c405ab 100644 --- a/python/nemo_relay/__init__.pyi +++ b/python/nemo_relay/__init__.pyi @@ -171,7 +171,8 @@ Return: JSON payload recorded on the emitted lifecycle event. Exceptional flow: - Exceptions fail open and preserve the last valid observability payload. + Exceptions fail closed, omit the observability payload, and stop the + remaining sanitizer chain. """ EventSanitizeGuardrail: TypeAlias = Callable[ [Event, EventSanitizeFields], @@ -186,7 +187,8 @@ Return: Observability fields recorded on the asynchronously published event. Exceptional flow: - Exceptions fail open and preserve the last valid event snapshot. + Exceptions fail closed, clear the mutable observability fields, and stop + the remaining sanitizer chain. """ ToolConditionalExecutionGuardrail: TypeAlias = Callable[[str, Json], Optional[str] | Awaitable[Optional[str]]] """Guardrail callback that can block tool execution. @@ -212,6 +214,10 @@ Arguments: Return: Request object recorded on the emitted lifecycle event, or ``None`` to omit the LLM observability payload and annotation. + +Exceptional flow: + Exceptions fail closed, omit the payload and annotation, and stop the + remaining sanitizer chain. """ LlmSanitizeResponseGuardrail: TypeAlias = Callable[ [Json, "LlmSanitizeResponseContext"], @@ -228,6 +234,10 @@ Arguments: Return: Response object recorded on the emitted lifecycle event, or ``None`` to omit the LLM observability payload and annotation. + +Exceptional flow: + Exceptions fail closed, omit the payload and annotation, and stop the + remaining sanitizer chain. """ LlmConditionalExecutionGuardrail: TypeAlias = Callable[[LLMRequest], Optional[str] | Awaitable[Optional[str]]] """Guardrail callback that can block an LLM call. diff --git a/python/nemo_relay/llm.py b/python/nemo_relay/llm.py index 96b2a8b55..de4c65ded 100644 --- a/python/nemo_relay/llm.py +++ b/python/nemo_relay/llm.py @@ -167,7 +167,8 @@ def call_end( ``call_end()`` remains synchronous. Sanitize-response guardrails, response-codec annotation, event sanitizers, and subscriber delivery run later on Relay's serial publication path. Callback and codec - failures are logged and fail open; they cannot be raised by this call. + sanitizer failures are logged and fail closed; they cannot be raised by + this call. Codec failures retain their documented fallback behavior. ``response_codec`` and ``annotated_response`` enrich observability output only and do not rewrite the caller-owned response. ``timestamp`` must be a timezone-aware ``datetime``; strings and naive diff --git a/python/nemo_relay/tools.py b/python/nemo_relay/tools.py index 1f53127c2..ea752ccdb 100644 --- a/python/nemo_relay/tools.py +++ b/python/nemo_relay/tools.py @@ -127,7 +127,7 @@ def call_end(handle, result, *, data=None, metadata=None, timestamp: datetime | Notes: ``call_end()`` remains synchronous. Sanitize-response guardrails, event sanitizers, and subscriber delivery run later on Relay's serial - publication path. Callback failures are logged and fail open; they + publication path. Callback failures are logged and fail closed; they cannot be raised by this call. The caller-owned ``result`` is not altered. ``timestamp`` must be a timezone-aware ``datetime``; strings and naive diff --git a/python/tests/test_event_sanitizers.py b/python/tests/test_event_sanitizers.py index 0c6cf56fb..b0fa2ed68 100644 --- a/python/tests/test_event_sanitizers.py +++ b/python/tests/test_event_sanitizers.py @@ -59,21 +59,33 @@ def second(event: nemo_relay.Event, fields: EventSanitizeFields) -> EventSanitiz assert calls == [("checkpoint", {"secret": "raw"}), ("mark", {"stage": "first"})] -def test_mark_sanitizer_exception_preserves_observability_fields(capture_events): +def test_mark_sanitizer_exception_clears_observability_fields(capture_events, capfd): _capture_name, events = capture_events + def seed_category_profile(_event: nemo_relay.Event, fields: EventSanitizeFields) -> EventSanitizeFields: + return { + "data": fields["data"], + "category_profile": {"subtype": "seeded"}, + "metadata": fields["metadata"], + } + def raises(_event: nemo_relay.Event, _fields: EventSanitizeFields) -> EventSanitizeFields: raise RuntimeError("sanitize boom") + guardrails.register_mark_sanitize("python-mark-seed-category-profile", 1, seed_category_profile) guardrails.register_mark_sanitize("python-mark-raises", 0, raises) try: + capfd.readouterr() scope.event("checkpoint", data={"kept": True}) subscribers.flush() finally: guardrails.deregister_mark_sanitize("python-mark-raises") + guardrails.deregister_mark_sanitize("python-mark-seed-category-profile") - assert events[-1].data == {"kept": True} + assert events[-1].data is None + assert events[-1].category_profile is None assert events[-1].metadata is None + assert "Python event sanitizer callable failed" in capfd.readouterr().err async def test_async_mark_sanitizer_runs_on_originating_loop(capture_events): diff --git a/python/tests/test_llm.py b/python/tests/test_llm.py index d45dc77fd..0065ed786 100644 --- a/python/tests/test_llm.py +++ b/python/tests/test_llm.py @@ -360,7 +360,7 @@ def test_duplicate_raises(self): guardrails.register_llm_sanitize_request("py_llm_dup", 1, lambda r, context: r) guardrails.deregister_llm_sanitize_request("py_llm_dup") - def test_sanitize_request_callable_error_preserves_observability_input(self): + def test_sanitize_request_callable_error_omits_observability_input(self): events = [] subscribers.register("py_llm_sanitize_req_sub", lambda event: events.append(event)) guardrails.register_llm_sanitize_request( @@ -383,10 +383,10 @@ def test_sanitize_request_callable_error_preserves_observability_input(self): subscribers.deregister("py_llm_sanitize_req_sub") start = _llm_event(events, "llm_sanitize_req_fail", "start") - assert start.data == {"headers": {"x-request-id": "safe"}, "content": request.content} + assert start.data is None assert start.annotated_request is None - def test_sanitize_request_invalid_return_preserves_observability_input(self): + def test_sanitize_request_invalid_return_omits_observability_input(self): events = [] subscribers.register("py_llm_sanitize_req_bad_sub", lambda event: events.append(event)) guardrails.register_llm_sanitize_request( @@ -409,10 +409,10 @@ def test_sanitize_request_invalid_return_preserves_observability_input(self): subscribers.deregister("py_llm_sanitize_req_bad_sub") start = _llm_event(events, "llm_sanitize_req_bad", "start") - assert start.data == {"headers": {"x-request-id": "safe"}, "content": request.content} + assert start.data is None assert start.annotated_request is None - def test_sanitize_response_callable_error_preserves_observability_output(self): + def test_sanitize_response_callable_error_omits_observability_output(self): events = [] subscribers.register("py_llm_sanitize_resp_sub", lambda event: events.append(event)) guardrails.register_llm_sanitize_response( @@ -431,10 +431,10 @@ def test_sanitize_response_callable_error_preserves_observability_output(self): subscribers.deregister("py_llm_sanitize_resp_sub") end = _llm_event(events, "llm_sanitize_resp_fail", "end") - assert end.data == {"ok": True} + assert end.data is None assert end.annotated_response is None - def test_sanitize_response_invalid_return_preserves_observability_output(self): + def test_sanitize_response_invalid_return_omits_observability_output(self): events = [] subscribers.register("py_llm_sanitize_resp_bad_sub", lambda event: events.append(event)) guardrails.register_llm_sanitize_response( @@ -453,7 +453,7 @@ def test_sanitize_response_invalid_return_preserves_observability_output(self): subscribers.deregister("py_llm_sanitize_resp_bad_sub") end = _llm_event(events, "llm_sanitize_resp_bad", "end") - assert end.data == {"ok": True} + assert end.data is None assert end.annotated_response is None def test_sanitize_response_guardrail_accepts_scalar_json_payloads(self): diff --git a/python/tests/test_tools.py b/python/tests/test_tools.py index a65ea43aa..51de3c645 100644 --- a/python/tests/test_tools.py +++ b/python/tests/test_tools.py @@ -276,7 +276,7 @@ def test_duplicate_guardrail_raises(self): guardrails.register_tool_sanitize_request("py_dup_guard", 1, lambda n, a: a) guardrails.deregister_tool_sanitize_request("py_dup_guard") - def test_sanitize_request_failure_falls_back_to_original_input(self): + def test_sanitize_request_failure_omits_observability_input(self): events = [] subscribers.register("py_tool_sanitize_req_sub", lambda event: events.append(event)) guardrails.register_tool_sanitize_request( @@ -293,9 +293,9 @@ def test_sanitize_request_failure_falls_back_to_original_input(self): subscribers.deregister("py_tool_sanitize_req_sub") start = _tool_event(events, "tool_sanitize_req_fail", "start") - assert start.data == {"value": 1} + assert start.data is None - def test_sanitize_response_invalid_return_falls_back_to_original_output(self): + def test_sanitize_response_invalid_return_omits_observability_output(self): events = [] subscribers.register("py_tool_sanitize_resp_sub", lambda event: events.append(event)) guardrails.register_tool_sanitize_response( @@ -312,7 +312,7 @@ def test_sanitize_response_invalid_return_falls_back_to_original_output(self): subscribers.deregister("py_tool_sanitize_resp_sub") end = _tool_event(events, "tool_sanitize_resp_bad", "end") - assert end.data == {"ok": True} + assert end.data is None def test_deregister_nonexistent(self): assert not guardrails.deregister_tool_sanitize_request("nonexistent")