From 20cf7f1bc5127b5bd082c53de3745556aacc8a83 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 13:31:57 -0400 Subject: [PATCH 1/9] feat: queue scope and mark event publication Signed-off-by: Will Killian --- .../src/api/runtime/subscriber_dispatcher.rs | 69 ++++++++++++++++++- crates/core/src/api/scope.rs | 59 ++++++++++------ crates/core/src/api/shared.rs | 27 ++++++-- .../subscriber_dispatcher_tests.rs | 65 +++++++++++++++++ docs/reference/event-sanitizers.mdx | 13 ++++ 5 files changed, 205 insertions(+), 28 deletions(-) diff --git a/crates/core/src/api/runtime/subscriber_dispatcher.rs b/crates/core/src/api/runtime/subscriber_dispatcher.rs index ffb4ef3b0..4ac440724 100644 --- a/crates/core/src/api/runtime/subscriber_dispatcher.rs +++ b/crates/core/src/api/runtime/subscriber_dispatcher.rs @@ -4,7 +4,10 @@ //! Asynchronous subscriber delivery for native targets. use crate::api::event::Event; -use crate::api::runtime::EventSubscriberFn; +use crate::api::registry::Guardrail; +use crate::api::runtime::{ + EventSanitizeFn, EventSubscriberFn, NemoRelayContextState, ScopeStackHandle, +}; use crate::error::Result; mod native { @@ -24,6 +27,7 @@ mod native { enum DispatcherMessage { Deliver { event: Box, + sanitizers: Vec>, subscribers: Vec, scope_stack: ScopeStackHandle, }, @@ -46,6 +50,7 @@ mod native { } let message = DispatcherMessage::Deliver { event: Box::new(event.clone()), + sanitizers: Vec::new(), subscribers: subscribers.to_vec(), scope_stack: current_scope_stack(), }; @@ -76,6 +81,41 @@ mod native { } } + pub(super) fn dispatch_sanitized_event( + event: Event, + sanitizers: Vec>, + subscribers: &[EventSubscriberFn], + scope_stack: ScopeStackHandle, + ) -> bool { + let message = DispatcherMessage::Deliver { + event: Box::new(event), + sanitizers, + subscribers: subscribers.to_vec(), + scope_stack, + }; + match dispatcher_sender() { + Ok(sender) if sender.send(message).is_ok() => true, + Ok(_) => { + log::warn!( + target: "nemo_relay.runtime", + event = "subscriber_event_dropped", + reason = "dispatcher_disconnected"; + "Subscriber event was dropped because the dispatcher stopped" + ); + false + } + Err(error) if !DISPATCHER_FAILURE_LOGGED.swap(true, Ordering::AcqRel) => { + log::error!( + target: "nemo_relay.runtime", + event = "subscriber_dispatcher_failed"; + "Subscriber dispatcher failed to start: {error}" + ); + false + } + Err(_) => false, + } + } + pub(super) fn flush_subscribers() -> Result<()> { if IN_DISPATCHER.with(Cell::get) { return Ok(()); @@ -149,9 +189,10 @@ mod native { match message { DispatcherMessage::Deliver { event, + sanitizers, subscribers, scope_stack, - } => deliver_event(event, subscribers, scope_stack), + } => deliver_event(event, sanitizers, subscribers, scope_stack), DispatcherMessage::Flush { done } => { let _ = done.send(()); } @@ -160,12 +201,25 @@ mod native { fn deliver_event( event: Box, + sanitizers: Vec>, subscribers: Vec, scope_stack: ScopeStackHandle, ) { let previous_scope_stack = capture_thread_scope_stack(); set_thread_scope_stack(scope_stack); IN_DISPATCHER.with(|flag| flag.set(true)); + let original = (*event).clone(); + let event = catch_unwind(AssertUnwindSafe(|| { + NemoRelayContextState::event_sanitize_snapshot_chain(*event, &sanitizers) + })) + .unwrap_or_else(|_| { + log::error!( + target: "nemo_relay.runtime", + event = "event_sanitizer_panicked"; + "Event sanitizer panicked; publishing the original event snapshot" + ); + original + }); for subscriber in subscribers { if catch_unwind(AssertUnwindSafe(|| subscriber(&event))).is_err() { log::error!( @@ -185,6 +239,17 @@ pub(crate) fn dispatch_event(event: &Event, subscribers: &[EventSubscriberFn]) - native::dispatch_event(event, subscribers) } +/// Queue a snapshot for serial event sanitization followed by subscriber +/// delivery. Used by synchronous scope and mark APIs. +pub(crate) fn dispatch_sanitized_event( + event: Event, + sanitizers: Vec>, + subscribers: &[EventSubscriberFn], + scope_stack: ScopeStackHandle, +) -> bool { + native::dispatch_sanitized_event(event, sanitizers, subscribers, scope_stack) +} + /// Wait for all queued subscriber callbacks submitted before this call. pub fn flush_subscribers() -> Result<()> { native::flush_subscribers() diff --git a/crates/core/src/api/scope.rs b/crates/core/src/api/scope.rs index 60a1aa53d..8cc0297e1 100644 --- a/crates/core/src/api/scope.rs +++ b/crates/core/src/api/scope.rs @@ -2,13 +2,14 @@ // SPDX-License-Identifier: Apache-2.0 use crate::api::event::{BaseEvent, CategoryProfile, DataSchema, EventCategory, MarkEvent}; -use crate::api::runtime::NemoRelayContextState; use crate::api::runtime::global_context; +use crate::api::runtime::subscriber_dispatcher; use crate::api::runtime::{ current_scope_stack, task_scope_push, task_scope_remove, task_scope_top, }; use crate::api::shared::{ - ensure_runtime_owner, resolve_parent_uuid, sanitize_event, snapshot_event_subscribers, + ensure_runtime_owner, resolve_parent_uuid, snapshot_event_sanitizers, + snapshot_event_subscribers, }; use crate::error::{FlowError, Result}; use crate::json::Json; @@ -216,12 +217,13 @@ pub fn get_handle() -> Result { /// cannot be read safely. /// /// # Notes -/// Scope-local subscribers attached to ancestor scopes observe the emitted -/// start event before the function returns. +/// The event and its visible middleware/subscriber chains are snapshotted +/// before this function returns. Sanitization and subscriber delivery happen +/// later on the serial publication dispatcher. pub fn push_scope(params: PushScopeParams<'_>) -> Result { ensure_runtime_owner()?; let parent_uuid = resolve_parent_uuid(params.parent); - let (handle, event, subscribers) = { + let (handle, event, subscribers, emission_scope_stack) = { let scope_stack = current_scope_stack(); let scope_guard = scope_stack.read().expect("scope stack lock poisoned"); let scope_subscribers = scope_guard.collect_scope_local_subscribers(); @@ -241,12 +243,16 @@ pub fn push_scope(params: PushScopeParams<'_>) -> Result { .build(); let handle = state.create_scope_handle(handle_params); let event = state.build_scope_start_event(&handle, params.input); - (handle, event, subscribers) + (handle, event, subscribers, scope_stack.clone()) }; - let event = sanitize_event(event); task_scope_push(handle.clone()); - if let Some(event) = event { - NemoRelayContextState::emit_event(&event, &subscribers); + if let Some(sanitizers) = snapshot_event_sanitizers(&event, &emission_scope_stack) { + let _ = subscriber_dispatcher::dispatch_sanitized_event( + event, + sanitizers, + &subscribers, + emission_scope_stack, + ); } Ok(handle) } @@ -276,7 +282,7 @@ pub fn push_scope(params: PushScopeParams<'_>) -> Result { pub fn pop_scope(params: PopScopeParams<'_>) -> Result<()> { ensure_runtime_owner()?; let scope_stack = current_scope_stack(); - let (scope, event, subscribers) = { + let (scope, event, subscribers, emission_scope_stack) = { let scope_guard = scope_stack.read().expect("scope stack lock poisoned"); let top = scope_guard.top(); if top.uuid != *params.handle_uuid { @@ -302,13 +308,20 @@ pub fn pop_scope(params: PopScopeParams<'_>) -> Result<()> { .metadata_opt(params.metadata) .build(), ); - (scope, event, subscribers) + (scope, event, subscribers, scope_stack.clone()) }; - let event = sanitize_event(event); + // Snapshot scope-local middleware before removing its owner. Publication + // happens later, but cleanup must not change the chain visible at emission. + let sanitizers = snapshot_event_sanitizers(&event, &emission_scope_stack); let removed = task_scope_remove(params.handle_uuid)?; debug_assert_eq!(removed.uuid, scope.uuid); - if let Some(event) = event { - NemoRelayContextState::emit_event(&event, &subscribers); + if let Some(sanitizers) = sanitizers { + let _ = subscriber_dispatcher::dispatch_sanitized_event( + event, + sanitizers, + &subscribers, + emission_scope_stack, + ); } Ok(()) } @@ -335,13 +348,14 @@ pub fn pop_scope(params: PopScopeParams<'_>) -> Result<()> { /// cannot be read safely. /// /// # Notes -/// Scope-local subscribers attached to ancestor scopes observe the emitted -/// mark event just like scope, tool, and LLM lifecycle events. +/// The event and its visible middleware/subscriber chains are snapshotted +/// before this function returns. Sanitization and subscriber delivery happen +/// later on the serial publication dispatcher. pub fn event(params: EmitMarkEventParams<'_>) -> Result<()> { ensure_runtime_owner()?; let parent_uuid = resolve_parent_uuid(params.parent); let scope_stack = current_scope_stack(); - let (event, subscribers) = { + let (event, subscribers, emission_scope_stack) = { let subscribers = if params.name == COMPACTION_EVENT_NAME { let mut scope_guard = scope_stack.write().expect("scope stack lock poisoned"); let subscribers = @@ -368,10 +382,15 @@ pub fn event(params: EmitMarkEventParams<'_>) -> Result<()> { params.category, params.category_profile, )); - (event, subscribers) + (event, subscribers, scope_stack.clone()) }; - if let Some(event) = sanitize_event(event) { - NemoRelayContextState::emit_event(&event, &subscribers); + if let Some(sanitizers) = snapshot_event_sanitizers(&event, &emission_scope_stack) { + let _ = subscriber_dispatcher::dispatch_sanitized_event( + event, + sanitizers, + &subscribers, + emission_scope_stack, + ); } Ok(()) } diff --git a/crates/core/src/api/shared.rs b/crates/core/src/api/shared.rs index a3baf051d..92aebeed1 100644 --- a/crates/core/src/api/shared.rs +++ b/crates/core/src/api/shared.rs @@ -7,8 +7,11 @@ use uuid::Uuid; use crate::api::event::{Event, ScopeCategory}; use crate::api::llm::LlmRequest; +use crate::api::registry::Guardrail; use crate::api::runtime::global_context; -use crate::api::runtime::{EventSubscriberFn, NemoRelayContextState, ScopeStackHandle}; +use crate::api::runtime::{ + EventSanitizeFn, EventSubscriberFn, NemoRelayContextState, ScopeStackHandle, +}; use crate::api::runtime::{current_scope_stack, task_scope_top}; use crate::api::scope::ScopeHandle; use crate::api::scope::ScopeType; @@ -51,7 +54,22 @@ pub(crate) fn sanitize_event_with_scope_stack( event: Event, scope_stack: &ScopeStackHandle, ) -> Option { - let entries = { + let entries = snapshot_event_sanitizers(&event, scope_stack)?; + Some(NemoRelayContextState::event_sanitize_snapshot_chain( + event, &entries, + )) +} + +/// Snapshot the event sanitizer chain visible on a captured scope stack. +/// +/// The snapshot remains valid after the emitting scope is removed, allowing +/// synchronous scope and mark APIs to enqueue publication without changing +/// which scope-local middleware observes the event. +pub(crate) fn snapshot_event_sanitizers( + event: &Event, + scope_stack: &ScopeStackHandle, +) -> Option>> { + Some({ let scope_guard = scope_stack.read().expect("scope stack lock poisoned"); let context = global_context(); let state = match context.read() { @@ -87,10 +105,7 @@ pub(crate) fn sanitize_event_with_scope_stack( ) } } - }; - Some(NemoRelayContextState::event_sanitize_snapshot_chain( - event, &entries, - )) + }) } pub(crate) fn ensure_runtime_owner() -> Result<()> { diff --git a/crates/core/tests/integration/subscriber_dispatcher_tests.rs b/crates/core/tests/integration/subscriber_dispatcher_tests.rs index e83fc7d6d..70fb5c87d 100644 --- a/crates/core/tests/integration/subscriber_dispatcher_tests.rs +++ b/crates/core/tests/integration/subscriber_dispatcher_tests.rs @@ -6,11 +6,16 @@ use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; +use nemo_relay::api::event::Event; +use nemo_relay::api::registry::{ + deregister_mark_sanitize_guardrail, register_mark_sanitize_guardrail, +}; use nemo_relay::api::runtime::{ NemoRelayContextState, create_scope_stack, global_context, set_thread_scope_stack, }; use nemo_relay::api::scope::{EmitMarkEventParams, event}; use nemo_relay::api::subscriber::{deregister_subscriber, flush_subscribers, register_subscriber}; +use serde_json::json; static TEST_MUTEX: Mutex<()> = Mutex::new(()); @@ -96,6 +101,66 @@ fn dispatcher_preserves_event_order() { assert_eq!(observed.lock().unwrap().as_slice(), ["one", "two"]); } +#[test] +fn mark_emission_snapshots_sanitizers_and_returns_before_they_finish() { + let _lock = TEST_MUTEX.lock().unwrap(); + flush_subscribers().unwrap(); + reset_global(); + setup_isolated_thread(); + + let (sanitizer_started_tx, sanitizer_started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let release_rx = Arc::new(Mutex::new(release_rx)); + register_mark_sanitize_guardrail( + "blocking-mark-sanitizer", + 10, + Arc::new(move |_, mut fields| { + sanitizer_started_tx.send(()).unwrap(); + release_rx.lock().unwrap().recv().unwrap(); + fields.data = Some(json!({"sanitized": true})); + fields + }), + ) + .unwrap(); + + let observed = Arc::new(Mutex::new(Vec::::new())); + let observed_events = Arc::clone(&observed); + register_subscriber( + "sanitized-mark-subscriber", + Arc::new(move |event| observed_events.lock().unwrap().push(event.clone())), + ) + .unwrap(); + + let (returned_tx, returned_rx) = mpsc::channel(); + let event_thread = std::thread::spawn(move || { + emit_mark("queued-sanitizer"); + returned_tx.send(()).unwrap(); + }); + + sanitizer_started_rx + .recv_timeout(Duration::from_secs(1)) + .expect("sanitizer should start on the dispatcher thread"); + returned_rx + .recv_timeout(Duration::from_secs(1)) + .expect("mark emission should return while its sanitizer is blocked"); + + // Removing the global registration cannot affect the already-snapshotted + // publication chain. + deregister_mark_sanitize_guardrail("blocking-mark-sanitizer").unwrap(); + release_tx.send(()).unwrap(); + event_thread.join().unwrap(); + flush_subscribers().unwrap(); + + let events = observed.lock().unwrap(); + assert_eq!(events.len(), 1); + assert_eq!( + events[0].sanitize_fields().data, + Some(json!({"sanitized": true})) + ); + drop(events); + deregister_subscriber("sanitized-mark-subscriber").unwrap(); +} + #[test] fn dispatcher_continues_after_subscriber_panic() { let _lock = TEST_MUTEX.lock().unwrap(); diff --git a/docs/reference/event-sanitizers.mdx b/docs/reference/event-sanitizers.mdx index 8ded00de5..8a4689662 100644 --- a/docs/reference/event-sanitizers.mdx +++ b/docs/reference/event-sanitizers.mdx @@ -53,6 +53,19 @@ 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()`. +## Publication Semantics + +Scope and mark emission APIs remain synchronous. They snapshot the event, +visible sanitizer chain, and subscribers, then enqueue that snapshot for +sanitization and publication on a serial background dispatcher. Subscribers +and exporters therefore receive the sanitized event after the emission call +returns. + +The dispatcher processes snapshots in FIFO order, preserving scope start/end +and mark ordering. Closing a scope or deregistering middleware after emission +does not alter an already-snapshotted publication chain. Use the binding's +subscriber flush API when a test or shutdown path must wait for queued delivery. + ## Registration Lifetimes Where you register a sanitizer determines how long it stays active. From 28ffebc95054cfbf16d0819b9fe7887b95eb9ca1 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 14:39:58 -0400 Subject: [PATCH 2/9] test: flush queued FFI event sanitizers Signed-off-by: Will Killian --- crates/ffi/tests/unit/api/registry_tests.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/ffi/tests/unit/api/registry_tests.rs b/crates/ffi/tests/unit/api/registry_tests.rs index 185cadd79..97452897c 100644 --- a/crates/ffi/tests/unit/api/registry_tests.rs +++ b/crates/ffi/tests/unit/api/registry_tests.rs @@ -243,6 +243,8 @@ fn test_ffi_event_sanitizer_registries_and_error_paths() { nemo_relay_deregister_mark_sanitize_guardrail(invalid_guard.as_ptr()), NemoRelayStatus::Ok ); + // The queued event retains its sanitizer snapshot after deregistration. + assert_eq!(nemo_relay_flush_subscribers(), NemoRelayStatus::Ok); assert_eq!(*lock_unpoisoned(plugin_frees()), 4); let mut owner = ptr::null_mut(); @@ -343,6 +345,8 @@ fn test_ffi_event_sanitizer_registries_and_error_paths() { ), NemoRelayStatus::Ok ); + // Scope removal does not alter the sanitizer snapshots already queued. + assert_eq!(nemo_relay_flush_subscribers(), NemoRelayStatus::Ok); assert_eq!(*lock_unpoisoned(plugin_frees()), 7); let invalid_uuid = cstring("not-a-uuid"); From 6f4b9ade43664f78dc2782b1f75c36d7b6d7a7ad Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 15:32:21 -0400 Subject: [PATCH 3/9] perf: skip queued sanitizers without subscribers Signed-off-by: Will Killian --- .../src/api/runtime/subscriber_dispatcher.rs | 3 +++ .../subscriber_dispatcher_tests.rs | 27 +++++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/crates/core/src/api/runtime/subscriber_dispatcher.rs b/crates/core/src/api/runtime/subscriber_dispatcher.rs index 4ac440724..39dea23fc 100644 --- a/crates/core/src/api/runtime/subscriber_dispatcher.rs +++ b/crates/core/src/api/runtime/subscriber_dispatcher.rs @@ -87,6 +87,9 @@ mod native { subscribers: &[EventSubscriberFn], scope_stack: ScopeStackHandle, ) -> bool { + if subscribers.is_empty() { + return true; + } let message = DispatcherMessage::Deliver { event: Box::new(event), sanitizers, diff --git a/crates/core/tests/integration/subscriber_dispatcher_tests.rs b/crates/core/tests/integration/subscriber_dispatcher_tests.rs index 70fb5c87d..037c05cc7 100644 --- a/crates/core/tests/integration/subscriber_dispatcher_tests.rs +++ b/crates/core/tests/integration/subscriber_dispatcher_tests.rs @@ -3,6 +3,7 @@ //! Integration tests for native subscriber dispatch behavior. +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; @@ -161,6 +162,32 @@ fn mark_emission_snapshots_sanitizers_and_returns_before_they_finish() { deregister_subscriber("sanitized-mark-subscriber").unwrap(); } +#[test] +fn mark_emission_skips_sanitizers_without_subscribers() { + let _lock = TEST_MUTEX.lock().unwrap(); + flush_subscribers().unwrap(); + reset_global(); + setup_isolated_thread(); + + let sanitizer_called = Arc::new(AtomicBool::new(false)); + let called = Arc::clone(&sanitizer_called); + register_mark_sanitize_guardrail( + "unused-mark-sanitizer", + 10, + Arc::new(move |_, fields| { + called.store(true, Ordering::Release); + fields + }), + ) + .unwrap(); + + emit_mark("no-subscribers"); + flush_subscribers().unwrap(); + deregister_mark_sanitize_guardrail("unused-mark-sanitizer").unwrap(); + + assert!(!sanitizer_called.load(Ordering::Acquire)); +} + #[test] fn dispatcher_continues_after_subscriber_panic() { let _lock = TEST_MUTEX.lock().unwrap(); From 452ef96243d3b7c6cb3f222acdac073e47625b1b Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 16:11:47 -0400 Subject: [PATCH 4/9] perf: avoid unsanitized event clones Signed-off-by: Will Killian --- .../src/api/runtime/subscriber_dispatcher.rs | 28 ++++++------ .../subscriber_dispatcher_tests.rs | 43 +++++++++++++++++++ 2 files changed, 59 insertions(+), 12 deletions(-) diff --git a/crates/core/src/api/runtime/subscriber_dispatcher.rs b/crates/core/src/api/runtime/subscriber_dispatcher.rs index 39dea23fc..29a18a525 100644 --- a/crates/core/src/api/runtime/subscriber_dispatcher.rs +++ b/crates/core/src/api/runtime/subscriber_dispatcher.rs @@ -211,18 +211,22 @@ mod native { let previous_scope_stack = capture_thread_scope_stack(); set_thread_scope_stack(scope_stack); IN_DISPATCHER.with(|flag| flag.set(true)); - let original = (*event).clone(); - let event = catch_unwind(AssertUnwindSafe(|| { - NemoRelayContextState::event_sanitize_snapshot_chain(*event, &sanitizers) - })) - .unwrap_or_else(|_| { - log::error!( - target: "nemo_relay.runtime", - event = "event_sanitizer_panicked"; - "Event sanitizer panicked; publishing the original event snapshot" - ); - original - }); + let event = if sanitizers.is_empty() { + *event + } else { + let original = (*event).clone(); + catch_unwind(AssertUnwindSafe(|| { + NemoRelayContextState::event_sanitize_snapshot_chain(*event, &sanitizers) + })) + .unwrap_or_else(|_| { + log::error!( + target: "nemo_relay.runtime", + event = "event_sanitizer_panicked"; + "Event sanitizer panicked; publishing the original event snapshot" + ); + original + }) + }; for subscriber in subscribers { if catch_unwind(AssertUnwindSafe(|| subscriber(&event))).is_err() { log::error!( diff --git a/crates/core/tests/integration/subscriber_dispatcher_tests.rs b/crates/core/tests/integration/subscriber_dispatcher_tests.rs index 037c05cc7..ead76744b 100644 --- a/crates/core/tests/integration/subscriber_dispatcher_tests.rs +++ b/crates/core/tests/integration/subscriber_dispatcher_tests.rs @@ -188,6 +188,49 @@ fn mark_emission_skips_sanitizers_without_subscribers() { assert!(!sanitizer_called.load(Ordering::Acquire)); } +#[test] +fn sanitizer_panic_publishes_the_original_event() { + let _lock = TEST_MUTEX.lock().unwrap(); + flush_subscribers().unwrap(); + reset_global(); + setup_isolated_thread(); + + register_mark_sanitize_guardrail( + "panicking-mark-sanitizer", + 10, + Arc::new(move |_, _| panic!("sanitizer failed")), + ) + .unwrap(); + + let observed = Arc::new(Mutex::new(Vec::::new())); + let observed_events = Arc::clone(&observed); + register_subscriber( + "panic-fallback-subscriber", + Arc::new(move |event| observed_events.lock().unwrap().push(event.clone())), + ) + .unwrap(); + + event( + EmitMarkEventParams::builder() + .name("panic-fallback") + .data(json!({"original": true})) + .build(), + ) + .unwrap(); + flush_subscribers().unwrap(); + + deregister_mark_sanitize_guardrail("panicking-mark-sanitizer").unwrap(); + deregister_subscriber("panic-fallback-subscriber").unwrap(); + + 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!({"original": true})) + ); +} + #[test] fn dispatcher_continues_after_subscriber_panic() { let _lock = TEST_MUTEX.lock().unwrap(); From fe6b5f0aa7ea53448740490c992ce7c3bd42ebf3 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 16:16:43 -0400 Subject: [PATCH 5/9] fix: preserve sanitized snapshots after callback panics Signed-off-by: Will Killian --- crates/core/src/api/runtime/state.rs | 16 ++++++++++++++-- .../src/api/runtime/subscriber_dispatcher.rs | 17 +---------------- .../integration/subscriber_dispatcher_tests.rs | 14 ++++++++++++-- 3 files changed, 27 insertions(+), 20 deletions(-) diff --git a/crates/core/src/api/runtime/state.rs b/crates/core/src/api/runtime/state.rs index eb3dbb950..fa45397a4 100644 --- a/crates/core/src/api/runtime/state.rs +++ b/crates/core/src/api/runtime/state.rs @@ -10,6 +10,7 @@ use std::any::Any; use std::collections::HashMap; +use std::panic::{AssertUnwindSafe, catch_unwind}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; @@ -638,8 +639,19 @@ impl NemoRelayContextState { entries: &[Guardrail], ) -> Event { for entry in entries { - let fields = (entry.payload)(&event, event.sanitize_fields()); - event.apply_sanitize_fields(fields); + if catch_unwind(AssertUnwindSafe(|| { + let fields = (entry.payload)(&event, event.sanitize_fields()); + event.apply_sanitize_fields(fields); + })) + .is_err() + { + log::error!( + target: "nemo_relay.runtime", + event = "event_sanitizer_panicked", + guardrail = entry.name.as_str(); + "Event sanitizer panicked; publishing the latest valid event snapshot" + ); + } } event } diff --git a/crates/core/src/api/runtime/subscriber_dispatcher.rs b/crates/core/src/api/runtime/subscriber_dispatcher.rs index 29a18a525..b050a6f45 100644 --- a/crates/core/src/api/runtime/subscriber_dispatcher.rs +++ b/crates/core/src/api/runtime/subscriber_dispatcher.rs @@ -211,22 +211,7 @@ mod native { let previous_scope_stack = capture_thread_scope_stack(); set_thread_scope_stack(scope_stack); IN_DISPATCHER.with(|flag| flag.set(true)); - let event = if sanitizers.is_empty() { - *event - } else { - let original = (*event).clone(); - catch_unwind(AssertUnwindSafe(|| { - NemoRelayContextState::event_sanitize_snapshot_chain(*event, &sanitizers) - })) - .unwrap_or_else(|_| { - log::error!( - target: "nemo_relay.runtime", - event = "event_sanitizer_panicked"; - "Event sanitizer panicked; publishing the original event snapshot" - ); - original - }) - }; + let event = NemoRelayContextState::event_sanitize_snapshot_chain(*event, &sanitizers); for subscriber in subscribers { if catch_unwind(AssertUnwindSafe(|| subscriber(&event))).is_err() { log::error!( diff --git a/crates/core/tests/integration/subscriber_dispatcher_tests.rs b/crates/core/tests/integration/subscriber_dispatcher_tests.rs index ead76744b..2276dee6c 100644 --- a/crates/core/tests/integration/subscriber_dispatcher_tests.rs +++ b/crates/core/tests/integration/subscriber_dispatcher_tests.rs @@ -189,12 +189,21 @@ fn mark_emission_skips_sanitizers_without_subscribers() { } #[test] -fn sanitizer_panic_publishes_the_original_event() { +fn sanitizer_panic_publishes_the_latest_valid_event() { let _lock = TEST_MUTEX.lock().unwrap(); flush_subscribers().unwrap(); reset_global(); setup_isolated_thread(); + register_mark_sanitize_guardrail( + "successful-mark-sanitizer", + 0, + Arc::new(move |_, mut fields| { + fields.data = Some(json!({"redacted": true})); + fields + }), + ) + .unwrap(); register_mark_sanitize_guardrail( "panicking-mark-sanitizer", 10, @@ -219,6 +228,7 @@ fn sanitizer_panic_publishes_the_original_event() { .unwrap(); flush_subscribers().unwrap(); + deregister_mark_sanitize_guardrail("successful-mark-sanitizer").unwrap(); deregister_mark_sanitize_guardrail("panicking-mark-sanitizer").unwrap(); deregister_subscriber("panic-fallback-subscriber").unwrap(); @@ -227,7 +237,7 @@ fn sanitizer_panic_publishes_the_original_event() { assert_eq!(events[0].name(), "panic-fallback"); assert_eq!( events[0].sanitize_fields().data, - Some(json!({"original": true})) + Some(json!({"redacted": true})) ); } From 7406c08143d0d57a55785dd40ba3e4f24197b3ad Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 18:08:05 -0400 Subject: [PATCH 6/9] docs: clarify deferred scope-end publication Signed-off-by: Will Killian --- crates/core/src/api/scope.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/core/src/api/scope.rs b/crates/core/src/api/scope.rs index 8cc0297e1..27763f836 100644 --- a/crates/core/src/api/scope.rs +++ b/crates/core/src/api/scope.rs @@ -279,6 +279,11 @@ pub fn push_scope(params: PushScopeParams<'_>) -> Result { /// /// # Notes /// The implicit root scope cannot be removed. +/// +/// Scope-end emission snapshots the visible scope-local sanitizers before +/// removing the scope. Publication is then queued after removal using that +/// snapshot, so cleanup does not change the middleware applied to the emitted +/// event. pub fn pop_scope(params: PopScopeParams<'_>) -> Result<()> { ensure_runtime_owner()?; let scope_stack = current_scope_stack(); From 4f9dc287ccf4e0fbf979837f75d17916c4d4b6bf Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 19:28:33 -0400 Subject: [PATCH 7/9] fix(node): avoid subscriber flush deadlock Signed-off-by: Will Killian --- crates/node/README.md | 9 ++++---- crates/node/src/api/mod.rs | 14 ++++++++----- crates/node/tests/event_sanitizers_tests.mjs | 22 ++++++++++---------- crates/node/tests/llm_tests.mjs | 2 +- crates/node/tests/scope_tests.mjs | 7 +++---- crates/node/tests/tools_tests.mjs | 2 +- 6 files changed, 30 insertions(+), 26 deletions(-) diff --git a/crates/node/README.md b/crates/node/README.md index 13d0ccda9..ea6486462 100644 --- a/crates/node/README.md +++ b/crates/node/README.md @@ -88,7 +88,7 @@ async function main() { event("initialized", handle, { binding: "node" }, null); }); - flushSubscribers(); + await flushSubscribers(); await new Promise((resolve) => setImmediate(resolve)); deregisterSubscriber("printer"); } @@ -99,9 +99,10 @@ main().catch((error) => { }); ``` -Native subscriber delivery is asynchronous. `flushSubscribers()` drains the -native dispatcher. The extra event-loop turn lets queued JavaScript callback -side effects complete before deregistration or exit. +Native subscriber delivery is asynchronous. Awaiting `flushSubscribers()` drains +the native dispatcher without blocking the Node.js event loop. The extra +event-loop turn lets queued JavaScript callback side effects complete before +deregistration or exit. The main runtime API is exported from `nemo-relay-node`. Additional entry points are available at `nemo-relay-node/typed`, `nemo-relay-node/plugin`, diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 24bc87b80..26838ee41 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -3237,17 +3237,21 @@ pub fn deregister_subscriber(name: String) -> Result { core_subscriber_api::deregister_subscriber(&name).map_err(to_napi_err) } -/// Wait for native subscriber callbacks queued before this call to finish. +/// Return a Promise that resolves when native subscriber callbacks queued +/// before this call finish. /// /// Call this function outside native subscriber callbacks. A re-entrant call returns without /// waiting to avoid blocking the dispatcher, so callbacks later in the same dispatch snapshot can /// still run. /// -/// JavaScript subscribers are queued through Node's `ThreadsafeFunction`; callers that -/// need JS callback side effects should await an event-loop tick after this returns. +/// JavaScript subscribers are queued through Node's `ThreadsafeFunction`. Awaiting this +/// Promise does not block the Node event loop while event sanitizers settle. #[napi] -pub fn flush_subscribers() -> Result<()> { - core_subscriber_api::flush_subscribers().map_err(to_napi_err) +pub async fn flush_subscribers() -> Result<()> { + tokio::task::spawn_blocking(core_subscriber_api::flush_subscribers) + .await + .map_err(|error| to_napi_err(FlowError::Internal(error.to_string())))? + .map_err(to_napi_err) } // --------------------------------------------------------------------------- diff --git a/crates/node/tests/event_sanitizers_tests.mjs b/crates/node/tests/event_sanitizers_tests.mjs index 812a2a0ef..3679b52cf 100644 --- a/crates/node/tests/event_sanitizers_tests.mjs +++ b/crates/node/tests/event_sanitizers_tests.mjs @@ -57,7 +57,7 @@ describe('event sanitizer registries', () => { }); try { lib.event('checkpoint', null, { secret: 'raw' }, { secret: 'raw' }); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 1); } finally { lib.deregisterMarkSanitizeGuardrail('node-event-first'); @@ -93,7 +93,7 @@ describe('event sanitizer registries', () => { { secret: 'input' }, ); lib.popScope(handle, { secret: 'output' }, null, { secret: 'end' }); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 2); } finally { lib.deregisterScopeSanitizeStartGuardrail('node-scope-start'); @@ -129,14 +129,14 @@ describe('event sanitizer registries', () => { lib.registerMarkSanitizeGuardrail(name, 0, sanitizer); try { lib.event(name, null, { kept: kind }, { kept: kind }); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, Object.keys(invalidResults).indexOf(kind) + 1); } finally { lib.deregisterMarkSanitizeGuardrail(seedName); lib.deregisterMarkSanitizeGuardrail(name); } assertSanitizerFieldsCleared(events.at(-1)); - assert.match(lib.getLastCallbackError(), /event sanitizer callback failed/); + assert.match(lib.getLastCallbackError(), /invalid JS event sanitizer result/); } } finally { lib.deregisterSubscriber('node-event-sanitize-invalid-sub'); @@ -151,7 +151,7 @@ describe('event sanitizer registries', () => { })); try { await lib.toolCallExecute('background-tool', { raw: true }, (args) => args); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 2); } finally { lib.deregisterScopeSanitizeStartGuardrail('node-background-start'); @@ -184,7 +184,7 @@ describe('event sanitizer registries', () => { lib.registerScopeSanitizeStartGuardrail(name, 0, sanitizer); try { await lib.toolCallExecute(name, { kept: kind }, (args) => args); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, (Object.keys(invalidResults).indexOf(kind) + 1) * 2); } finally { lib.deregisterScopeSanitizeStartGuardrail(seedName); @@ -215,7 +215,7 @@ describe('event sanitizer registries', () => { }); try { await lib.toolCallExecute('background-throw-tool', { kept: true }, (args) => args); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 2); const start = events.find( (event) => event.kind === 'scope' && event.name === 'background-throw-tool' && event.scope_category === 'start', @@ -243,7 +243,7 @@ describe('event sanitizer registries', () => { lib.popScope(child); lib.popScope(owner); lib.event('outside', null, { raw: true }); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 3); lib.deregisterSubscriber('node-event-sanitize-local-sub'); const marks = Object.fromEntries( @@ -271,11 +271,11 @@ describe('event sanitizer registries', () => { components: [plugin.ComponentSpec(kind)], }); lib.event('configured', null, { raw: true }); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 1); plugin.clear(); lib.event('cleared', null, { raw: true }); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 2); } finally { plugin.clear(); @@ -312,7 +312,7 @@ describe('event sanitizer registries', () => { components: [plugin.ComponentSpec(kind)], }); lib.event('plugin-throw', null, { raw: true }, { raw: true }); - lib.flushSubscribers(); + await lib.flushSubscribers(); await waitFor(events, 1); assertSanitizerFieldsCleared(events.at(-1)); assert.match(lib.getLastCallbackError() ?? '', /plugin sanitizer boom/i); diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index 3115b2136..61eabb89b 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/crates/node/tests/llm_tests.mjs @@ -51,7 +51,7 @@ function rejectWith(value) { } async function flushSubscriberCallbacks() { - flushSubscribers(); + await flushSubscribers(); for (let i = 0; i < 10; i += 1) { await new Promise((resolve) => setImmediate(resolve)); } diff --git a/crates/node/tests/scope_tests.mjs b/crates/node/tests/scope_tests.mjs index a4cb41aed..cb50f1ade 100644 --- a/crates/node/tests/scope_tests.mjs +++ b/crates/node/tests/scope_tests.mjs @@ -30,7 +30,7 @@ function rejectWithPrimitive(value) { } async function flushSubscriberCallbacks() { - flushSubscribers(); + await flushSubscribers(); for (let i = 0; i < 10; i += 1) { await new Promise((resolve) => setImmediate(resolve)); } @@ -362,13 +362,12 @@ describe('Subscribers', () => { } }); - it('flushSubscribers is a native barrier before JS event-loop delivery', async () => { + it('flushSubscribers asynchronously drains the native dispatcher', async () => { const events = []; registerSubscriber('node_flush_collector', (e) => events.push(e)); try { event('node_flush_mark', null, null, null); - flushSubscribers(); - assert.equal(events.length, 0); + await flushSubscribers(); await new Promise((resolve) => setImmediate(resolve)); assert.ok(events.some((e) => e.kind === 'mark' && e.name === 'node_flush_mark')); } finally { diff --git a/crates/node/tests/tools_tests.mjs b/crates/node/tests/tools_tests.mjs index 125475561..3d20628b6 100644 --- a/crates/node/tests/tools_tests.mjs +++ b/crates/node/tests/tools_tests.mjs @@ -48,7 +48,7 @@ function sparseArray() { } async function waitForSubscriberCallbacks(predicate, timeoutMs = 15000) { - flushSubscribers(); + await flushSubscribers(); // flushSubscribers() waits for Relay's Rust subscriber dispatcher, but JS // subscriber callbacks are queued onto Node's event loop through N-API // ThreadsafeFunction. Yield event-loop turns until the observed JS-side From 44f8ef030ed2896714df3b10455f680dd80f1787 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 28 Jul 2026 19:56:31 -0400 Subject: [PATCH 8/9] fix(node): await subscriber flush in OpenClaw Signed-off-by: Will Killian --- crates/node/src/api/mod.rs | 3 +++ integrations/openclaw/src/hooks-backend.ts | 6 +++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 26838ee41..24df871fb 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -3246,6 +3246,9 @@ pub fn deregister_subscriber(name: String) -> Result { /// /// JavaScript subscribers are queued through Node's `ThreadsafeFunction`. Awaiting this /// Promise does not block the Node event loop while event sanitizers settle. +/// +/// The Promise rejects if the blocking task fails or the core subscriber flush returns an error. +/// Callers should handle errors when awaiting it. #[napi] pub async fn flush_subscribers() -> Result<()> { tokio::task::spawn_blocking(core_subscriber_api::flush_subscribers) diff --git a/integrations/openclaw/src/hooks-backend.ts b/integrations/openclaw/src/hooks-backend.ts index e72f5c6cb..41bcca99b 100644 --- a/integrations/openclaw/src/hooks-backend.ts +++ b/integrations/openclaw/src/hooks-backend.ts @@ -426,7 +426,7 @@ export class HookReplayBackend { this.materializeDeferredSessionRoot(session); drainSession(this.sessionManager(), session); closeSessionRoot(this.sessionManager(), session, summary, session.finalOutput ?? summary, metadata); - this.flushSubscriberDelivery('session_close'); + await this.flushSubscriberDelivery('session_close'); this.forgetPendingSubagentLineage(session); deleteSession(this.stateValue, session); } @@ -466,9 +466,9 @@ export class HookReplayBackend { } /** Wait for native subscriber/exporter delivery after a replay closure boundary. */ - private flushSubscriberDelivery(label: string): void { + private async flushSubscriberDelivery(label: string): Promise { try { - this.nf.flushSubscribers?.(); + await this.nf.flushSubscribers?.(); } catch (error) { this.logBoundedWarn( `flush-subscribers:${label}`, From f5b94493c1b1b10e4f2a7785c2b1f31bf7488043 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Wed, 29 Jul 2026 14:28:19 -0400 Subject: [PATCH 9/9] test(ffi): make OTLP collector socket blocking Signed-off-by: Will Killian --- crates/ffi/tests/unit/api/registry_tests.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/ffi/tests/unit/api/registry_tests.rs b/crates/ffi/tests/unit/api/registry_tests.rs index 97452897c..7c3af6ee5 100644 --- a/crates/ffi/tests/unit/api/registry_tests.rs +++ b/crates/ffi/tests/unit/api/registry_tests.rs @@ -22,6 +22,7 @@ fn start_otlp_http_collector() -> (String, Receiver>, JoinHandle<()>) { while Instant::now() < deadline { match listener.accept() { Ok((mut stream, _)) => { + stream.set_nonblocking(false).unwrap(); stream .set_read_timeout(Some(Duration::from_secs(1))) .unwrap();