feat: queue scope and mark event publication

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<EventSanitizeFn>],
     ) -> 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
     }

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<Event>,
+            sanitizers: Vec<Guardrail<EventSanitizeFn>>,
             subscribers: Vec<EventSubscriberFn>,
             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,44 @@ mod native {
         }
     }
 
+    pub(super) fn dispatch_sanitized_event(
+        event: Event,
+        sanitizers: Vec<Guardrail<EventSanitizeFn>>,
+        subscribers: &[EventSubscriberFn],
+        scope_stack: ScopeStackHandle,
+    ) -> bool {
+        if subscribers.is_empty() {
+            return true;
+        }
+        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 +192,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 +204,14 @@ mod native {
 
     fn deliver_event(
         event: Box<Event>,
+        sanitizers: Vec<Guardrail<EventSanitizeFn>>,
         subscribers: Vec<EventSubscriberFn>,
         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 event = NemoRelayContextState::event_sanitize_snapshot_chain(*event, &sanitizers);
         for subscriber in subscribers {
             if catch_unwind(AssertUnwindSafe(|| subscriber(&event))).is_err() {
                 log::error!(
@@ -185,6 +231,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<Guardrail<EventSanitizeFn>>,
+    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()

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<ScopeHandle> {
 /// 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<ScopeHandle> {
     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<ScopeHandle> {
             .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)
 }
@@ -273,10 +279,15 @@ pub fn push_scope(params: PushScopeParams<'_>) -> Result<ScopeHandle> {
 ///
 /// # 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();
-    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 +313,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 +353,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 +387,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(())
 }

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<Event> {
-    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<Vec<Guardrail<EventSanitizeFn>>> {
+    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<()> {