diff --git a/crates/node/pii_redaction.d.ts b/crates/node/pii_redaction.d.ts index e4a9f9f2d..5a5f20e76 100644 --- a/crates/node/pii_redaction.d.ts +++ b/crates/node/pii_redaction.d.ts @@ -37,6 +37,7 @@ export interface Config { output?: boolean; tool_input?: boolean; tool_output?: boolean; + mark?: boolean; priority?: number; codec?: 'openai_chat' | 'openai_responses' | 'anthropic_messages' | string; builtin?: BuiltinConfig; diff --git a/crates/node/pii_redaction.js b/crates/node/pii_redaction.js index 941bde9bb..c5fc6d313 100644 --- a/crates/node/pii_redaction.js +++ b/crates/node/pii_redaction.js @@ -20,6 +20,7 @@ function defaultConfig() { output: true, tool_input: true, tool_output: true, + mark: true, priority: 100, }; } diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index b8f199bf5..76a84ee28 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -import type { Json } from './index'; +import type { EventSanitizeFields, Json } from './index'; /** Policy behavior for unsupported configuration. */ export type UnsupportedBehavior = 'ignore' | 'warn' | 'error'; @@ -140,6 +140,24 @@ export interface ToolExecutionInterceptOutcome { export interface PluginContext { /** Register an infallible event subscriber for this component. */ registerSubscriber(name: string, callback: (event: Json) => void): void; + /** Register a mark event sanitizer for this component. */ + registerMarkSanitizeGuardrail( + name: string, + priority: number, + callback: (event: Json, fields: EventSanitizeFields) => EventSanitizeFields, + ): void; + /** Register a scope-start event sanitizer for this component. */ + registerScopeSanitizeStartGuardrail( + name: string, + priority: number, + callback: (event: Json, fields: EventSanitizeFields) => EventSanitizeFields, + ): void; + /** Register a scope-end event sanitizer for this component. */ + registerScopeSanitizeEndGuardrail( + name: string, + priority: number, + callback: (event: Json, fields: EventSanitizeFields) => EventSanitizeFields, + ): void; /** Register a tool sanitize-request guardrail for this component. */ registerToolSanitizeRequestGuardrail( name: string, diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index fa30cd609..9f06b649f 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -30,7 +30,9 @@ use tokio_stream::StreamExt; use nemo_relay::api::llm as core_llm_api; use nemo_relay::api::llm::{LlmAttributes, LlmRequest}; use nemo_relay::api::registry as core_registry_api; -use nemo_relay::api::runtime::{LlmExecutionNextFn, LlmStreamExecutionNextFn, ToolExecutionNextFn}; +use nemo_relay::api::runtime::{ + EventSanitizeFn, LlmExecutionNextFn, LlmStreamExecutionNextFn, ToolExecutionNextFn, +}; use nemo_relay::api::runtime::{ TASK_SCOPE_STACK, create_scope_stack as create_scope_stack_handle, current_scope_stack as current_scope_stack_handle, scope_stack_active as scope_stack_is_active, @@ -65,10 +67,13 @@ use crate::callable; use crate::convert::{ callback_json, clear_last_callback_error as clear_recorded_callback_error, get_last_callback_error as get_recorded_callback_error, opt_json, parse_timestamp_micros, - to_napi_err, + record_callback_error, to_napi_err, }; use crate::stream::LlmStream; -use crate::types::{LlmHandle, ScopeHandle, ScopeStack, ScopeType, ToolHandle}; +use crate::types::{ + EventSanitizeFields, LlmHandle, ScopeHandle, ScopeStack, ScopeType, ToolHandle, + event_sanitize_fields_from_js, +}; #[napi::module_init] fn init() { @@ -421,6 +426,40 @@ fn json_callback_tsfn( Ok(tsfn) } +#[allow(clippy::too_many_arguments)] +fn add_plugin_event_sanitizer( + env: &Env, + context: &mut JsObject, + property: &str, + namespace_prefix: String, + registrations: Arc>>, + register: fn(&str, i32, EventSanitizeFn) -> FlowResult<()>, + deregister: fn(&str) -> FlowResult, + label: &'static str, +) -> napi::Result<()> { + let function = env.create_function_from_closure(property, move |ctx| { + let name = format!("{}{}", namespace_prefix, ctx.get::(0)?); + let priority = ctx.get::(1)?; + let callback = ctx.get::(2)?; + register(&name, priority, node_event_sanitize_fn(ctx.env, &callback)?) + .map_err(to_napi_err)?; + let name_clone = name.clone(); + registrations.lock().unwrap().push(PluginRegistration::new( + "plugin", + name_clone.clone(), + Box::new(move || { + deregister(&name_clone).map(|_| ()).map_err(|error| { + PluginError::RegistrationFailed(format!( + "{label} deregistration failed: {error}" + )) + }) + }), + )); + ctx.env.get_undefined() + })?; + context.set_named_property(property, function) +} + fn build_plugin_context( env: &Env, namespace_prefix: String, @@ -464,6 +503,37 @@ fn build_plugin_context( )?; context.set_named_property("registerSubscriber", register_subscriber)?; + add_plugin_event_sanitizer( + env, + &mut context, + "registerMarkSanitizeGuardrail", + namespace_prefix.clone(), + registrations.clone(), + core_registry_api::register_mark_sanitize_guardrail, + core_registry_api::deregister_mark_sanitize_guardrail, + "mark sanitize guardrail", + )?; + add_plugin_event_sanitizer( + env, + &mut context, + "registerScopeSanitizeStartGuardrail", + namespace_prefix.clone(), + registrations.clone(), + core_registry_api::register_scope_sanitize_start_guardrail, + core_registry_api::deregister_scope_sanitize_start_guardrail, + "scope start sanitize guardrail", + )?; + add_plugin_event_sanitizer( + env, + &mut context, + "registerScopeSanitizeEndGuardrail", + namespace_prefix.clone(), + registrations.clone(), + core_registry_api::register_scope_sanitize_end_guardrail, + core_registry_api::deregister_scope_sanitize_end_guardrail, + "scope end sanitize guardrail", + )?; + let tool_sanitize_request_regs = registrations.clone(); let tool_sanitize_request_namespace = namespace_prefix.clone(); let register_tool_sanitize_request_guardrail = env.create_function_from_closure( @@ -1021,6 +1091,104 @@ impl PersistentJsFunction { // environment stored on this struct. unsafe { Option::::from_napi_value(self.env, returned.raw()) }.map(callback_json) } + + fn call_event_sanitize( + &self, + event: Json, + fields: EventSanitizeFields, + ) -> napi::Result { + let mut value = ptr::null_mut(); + // SAFETY: `self.reference` is a live N-API reference created in + // `self.env`, and `value` is writable storage for the borrowed + // function value. + let status = + unsafe { napi::sys::napi_get_reference_value(self.env, self.reference, &mut value) }; + if status != napi::sys::Status::napi_ok { + return Err(napi::Error::from_reason( + "failed to borrow event sanitizer function", + )); + } + // SAFETY: `value` was resolved from this struct's function reference, + // so it is a live function value in `self.env` for this call. + let func = unsafe { JsFunction::from_raw_unchecked(self.env, value) }; + // SAFETY: `Json::to_napi_value` created this event value in `self.env`, + // so wrapping it as `JsUnknown` is valid for the immediate callback. + let event = unsafe { + JsUnknown::from_raw_unchecked(self.env, Json::to_napi_value(self.env, event)?) + }; + // SAFETY: `EventSanitizeFields::to_napi_value` created this fields + // value in `self.env`, so wrapping it as `JsUnknown` is valid for the + // immediate callback. + let fields = unsafe { + JsUnknown::from_raw_unchecked( + self.env, + EventSanitizeFields::to_napi_value(self.env, fields)?, + ) + }; + let returned = func.call(None, &[event, fields])?; + event_sanitize_fields_from_js(returned) + } +} + +fn core_event_fields( + fields: EventSanitizeFields, +) -> Option { + Some(nemo_relay::api::event::EventSanitizeFields { + data: fields.data, + category_profile: fields + .category_profile + .map(serde_json::from_value) + .transpose() + .ok()?, + metadata: fields.metadata, + }) +} + +fn js_event_fields(fields: &nemo_relay::api::event::EventSanitizeFields) -> EventSanitizeFields { + EventSanitizeFields { + data: fields.data.clone(), + category_profile: fields + .category_profile + .as_ref() + .and_then(|value| serde_json::to_value(value).ok()), + metadata: fields.metadata.clone(), + } +} + +fn node_event_sanitize_fn(env: &Env, func: &JsFunction) -> napi::Result { + let direct = Arc::new(PersistentJsFunction::new(env, func)?); + let register_thread = std::thread::current().id(); + let mut tsfn = func.create_threadsafe_function( + 0, + |ctx: napi::threadsafe_function::ThreadSafeCallContext<(Json, Json)>| { + Ok(vec![ctx.value.0, ctx.value.1]) + }, + )?; + tsfn.unref(env)?; + let background = callable::wrap_js_event_sanitize_fn(tsfn); + Ok(Arc::new(move |event, fields| { + if std::thread::current().id() == register_thread { + let event_json = match event.try_to_json_value() { + Ok(event_json) => event_json, + Err(error) => { + record_callback_error(format!( + "nemo_relay: failed to serialize JS event sanitizer context: {error}" + )); + return fields; + } + }; + let sanitized = direct + .call_event_sanitize(event_json, js_event_fields(&fields)) + .ok() + .and_then(core_event_fields); + if sanitized.is_none() { + record_callback_error("nemo_relay: JS event sanitizer callback failed".to_string()); + } + sanitized.unwrap_or(fields) + } else { + background(event, fields) + } + })) } impl Drop for PersistentJsFunction { @@ -2000,6 +2168,48 @@ pub fn llm_stream_call_execute( // Tool guardrail registrations // --------------------------------------------------------------------------- +macro_rules! napi_event_guardrail_api { + ($register_name:ident, $deregister_name:ident, $core_register:path, $core_deregister:path) => { + #[napi] + pub fn $register_name( + env: Env, + name: String, + priority: i32, + #[napi( + ts_arg_type = "(event: Json, fields: EventSanitizeFields) => EventSanitizeFields" + )] + guardrail: JsFunction, + ) -> Result<()> { + $core_register(&name, priority, node_event_sanitize_fn(&env, &guardrail)?) + .map_err(to_napi_err) + } + + #[napi] + pub fn $deregister_name(name: String) -> Result { + $core_deregister(&name).map_err(to_napi_err) + } + }; +} + +napi_event_guardrail_api!( + register_mark_sanitize_guardrail, + deregister_mark_sanitize_guardrail, + core_registry_api::register_mark_sanitize_guardrail, + core_registry_api::deregister_mark_sanitize_guardrail +); +napi_event_guardrail_api!( + register_scope_sanitize_start_guardrail, + deregister_scope_sanitize_start_guardrail, + core_registry_api::register_scope_sanitize_start_guardrail, + core_registry_api::deregister_scope_sanitize_start_guardrail +); +napi_event_guardrail_api!( + register_scope_sanitize_end_guardrail, + deregister_scope_sanitize_end_guardrail, + core_registry_api::register_scope_sanitize_end_guardrail, + core_registry_api::deregister_scope_sanitize_end_guardrail +); + macro_rules! napi_guardrail_tool_api { ($(#[doc = $reg_doc:expr_2021])* $register_name:ident, $(#[doc = $dereg_doc:expr_2021])* $deregister_name:ident, @@ -2409,6 +2619,58 @@ pub fn flush_subscribers() -> Result<()> { // Scope-local guardrail registrations — Tool // --------------------------------------------------------------------------- +macro_rules! napi_scope_event_guardrail_api { + ($register_name:ident, $deregister_name:ident, $core_register:path, $core_deregister:path) => { + #[napi] + pub fn $register_name( + env: Env, + scope_uuid: String, + name: String, + priority: i32, + #[napi( + ts_arg_type = "(event: Json, fields: EventSanitizeFields) => EventSanitizeFields" + )] + guardrail: JsFunction, + ) -> Result<()> { + let uuid = uuid::Uuid::parse_str(&scope_uuid) + .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; + $core_register( + &uuid, + &name, + priority, + node_event_sanitize_fn(&env, &guardrail)?, + ) + .map_err(to_napi_err) + } + + #[napi] + pub fn $deregister_name(scope_uuid: String, name: String) -> Result { + let uuid = uuid::Uuid::parse_str(&scope_uuid) + .map_err(|e| napi::Error::from_reason(format!("invalid UUID: {e}")))?; + $core_deregister(&uuid, &name).map_err(to_napi_err) + } + }; +} + +napi_scope_event_guardrail_api!( + scope_register_mark_sanitize_guardrail, + scope_deregister_mark_sanitize_guardrail, + core_registry_api::scope_register_mark_sanitize_guardrail, + core_registry_api::scope_deregister_mark_sanitize_guardrail +); +napi_scope_event_guardrail_api!( + scope_register_scope_sanitize_start_guardrail, + scope_deregister_scope_sanitize_start_guardrail, + core_registry_api::scope_register_scope_sanitize_start_guardrail, + core_registry_api::scope_deregister_scope_sanitize_start_guardrail +); +napi_scope_event_guardrail_api!( + scope_register_scope_sanitize_end_guardrail, + scope_deregister_scope_sanitize_end_guardrail, + core_registry_api::scope_register_scope_sanitize_end_guardrail, + core_registry_api::scope_deregister_scope_sanitize_end_guardrail +); + macro_rules! napi_scope_guardrail_tool_api { ($(#[doc = $reg_doc:expr_2021])* $register_name:ident, $(#[doc = $dereg_doc:expr_2021])* $deregister_name:ident, diff --git a/crates/node/src/callable.rs b/crates/node/src/callable.rs index 29fb61644..8c5853bab 100644 --- a/crates/node/src/callable.rs +++ b/crates/node/src/callable.rs @@ -15,15 +15,18 @@ use std::sync::Arc; use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode}; use nemo_relay::api::runtime::{ - EventSubscriberFn, LlmConditionalFn, LlmExecutionNextFn, LlmRequestInterceptFn, - LlmSanitizeRequestFn, LlmSanitizeResponseFn, LlmStreamExecutionNextFn, ToolConditionalFn, - ToolExecutionNextFn, ToolInterceptFn, ToolSanitizeFn, + EventSanitizeFn, EventSubscriberFn, LlmConditionalFn, LlmExecutionNextFn, + LlmRequestInterceptFn, LlmSanitizeRequestFn, LlmSanitizeResponseFn, LlmStreamExecutionNextFn, + ToolConditionalFn, ToolExecutionNextFn, ToolInterceptFn, ToolSanitizeFn, }; use serde::{Deserialize, Serialize}; use serde_json::Value as Json; use tokio_stream::StreamExt; -use nemo_relay::api::event::{CategoryProfile, Event, EventCategory, PendingMarkSpec}; +use nemo_relay::api::event::{ + CategoryProfile, Event, EventCategory, EventSanitizeFields as CoreEventSanitizeFields, + PendingMarkSpec, +}; use nemo_relay::api::llm::{LlmRequest, LlmRequestInterceptOutcome}; use nemo_relay::api::tool::ToolExecutionInterceptOutcome; use nemo_relay::codec::optimization::LlmOptimizationContribution; @@ -34,7 +37,7 @@ use nemo_relay::error::{FlowError, Result}; use crate::convert::{callback_json, record_callback_error}; use crate::promise_call::{JsonNextFn, JsonStreamNextFn, PromiseAwareFn}; -use crate::types::JsEvent; +use crate::types::{EventSanitizeFields, JsEvent, event_sanitize_fields_from_js}; /// JavaScript-facing pending mark DTO. #[derive(Debug, Deserialize, Serialize)] @@ -514,6 +517,74 @@ pub fn wrap_js_event_subscriber( }) } +/// Wrap a JS event sanitizer: ``(event, fields) => fields``. +pub fn wrap_js_event_sanitize_fn( + func: ThreadsafeFunction<(Json, Json), ErrorStrategy::Fatal>, +) -> EventSanitizeFn { + let func = Arc::new(func); + Arc::new(move |event: &Event, fields: CoreEventSanitizeFields| { + let event_json = match JsEvent::try_from_event(event) { + Ok(event) => event.into_json(), + Err(error) => { + record_callback_error(format!( + "nemo_relay: failed to serialize JS event sanitizer context: {error}" + )); + return fields.clone(); + } + }; + let js_fields = EventSanitizeFields { + data: fields.data.clone(), + category_profile: fields + .category_profile + .as_ref() + .and_then(|value| serde_json::to_value(value).ok()), + metadata: fields.metadata.clone(), + }; + let (tx, rx) = std::sync::mpsc::channel(); + let status = func.call_with_return_value( + ( + event_json, + serde_json::to_value(js_fields).unwrap_or(Json::Null), + ), + ThreadsafeFunctionCallMode::Blocking, + move |value: napi::JsUnknown| { + let _ = tx.send(event_sanitize_fields_from_js(value)); + Ok(()) + }, + ); + if status != napi::Status::Ok { + record_callback_error(format!( + "nemo_relay: failed to queue JS event sanitizer callback: {status:?}" + )); + return fields.clone(); + } + rx.recv() + .map_err(|error| { + record_callback_error(format!( + "nemo_relay: JS event sanitizer callback failed: {error}" + )); + }) + .ok() + .and_then(|result| result.ok()) + .and_then(|result| { + let category_profile = result + .category_profile + .map(serde_json::from_value) + .transpose() + .ok()?; + Some(CoreEventSanitizeFields { + data: result.data, + category_profile, + metadata: result.metadata, + }) + }) + .unwrap_or_else(|| { + record_callback_error("nemo_relay: invalid JS event sanitizer result".to_string()); + fields.clone() + }) + }) +} + // --------------------------------------------------------------------------- // Codec wrappers // --------------------------------------------------------------------------- diff --git a/crates/node/src/types/mod.rs b/crates/node/src/types/mod.rs index 58cb37c04..6f5f96fe6 100644 --- a/crates/node/src/types/mod.rs +++ b/crates/node/src/types/mod.rs @@ -7,9 +7,10 @@ //! and attribute constants that are exposed to JavaScript/TypeScript consumers. //! Doc comments on `#[napi]` items are emitted into the generated `index.d.ts`. +use napi::{JsObject, JsUnknown}; use napi_derive::napi; use nemo_relay::api::runtime::{ScopeStackHandle, create_scope_stack}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use serde_json::Value as Json; use nemo_relay::api::event::Event; @@ -299,6 +300,57 @@ impl LlmRequest { // Event (read-only, for subscribers) // --------------------------------------------------------------------------- +/// Observability fields returned by mark and scope event sanitizers. +#[napi(object)] +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct EventSanitizeFields { + pub data: Option, + pub category_profile: Option, + pub metadata: Option, +} + +pub(crate) fn event_sanitize_fields_from_js(value: JsUnknown) -> napi::Result { + let object = match value.get_type()? { + napi::ValueType::Object => JsObject::try_from(value)?, + _ => { + return Err(napi::Error::from_reason( + "event sanitizer must return an object", + )); + } + }; + if object.is_array()? || object.has_named_property("then")? { + return Err(napi::Error::from_reason( + "event sanitizer must return a non-thenable object", + )); + } + + let mut has_field = false; + for field in ["data", "categoryProfile", "metadata"] { + has_field |= object.has_own_property(field)?; + } + if !has_field { + return Err(napi::Error::from_reason( + "event sanitizer must return at least one sanitizer field", + )); + } + + let field = |name| { + if object.has_own_property(name)? { + object.get(name) + } else { + Ok(None) + } + }; + Ok(EventSanitizeFields { + data: field("data")?, + category_profile: field("categoryProfile")?, + metadata: field("metadata")?, + }) +} + +// --------------------------------------------------------------------------- + /// A read-only ATOF lifecycle event delivered to subscribers. #[derive(Serialize)] #[serde(transparent)] diff --git a/crates/node/tests/adaptive_tests.mjs b/crates/node/tests/adaptive_tests.mjs index 540590fa2..960ecb2c6 100644 --- a/crates/node/tests/adaptive_tests.mjs +++ b/crates/node/tests/adaptive_tests.mjs @@ -113,6 +113,9 @@ describe('core plugins', () => { hasToolRequest: typeof context.registerToolRequestIntercept === 'function', hasLlmExecution: typeof context.registerLlmExecutionIntercept === 'function', hasLlmStreamExecution: typeof context.registerLlmStreamExecutionIntercept === 'function', + hasMarkSanitize: typeof context.registerMarkSanitizeGuardrail === 'function', + hasScopeStartSanitize: typeof context.registerScopeSanitizeStartGuardrail === 'function', + hasScopeEndSanitize: typeof context.registerScopeSanitizeEndGuardrail === 'function', }; context.registerSubscriber('subscriber', () => {}); context.registerToolRequestIntercept('toolRequest', 17, false, (_name, args) => ({ @@ -154,6 +157,9 @@ describe('core plugins', () => { hasToolRequest: true, hasLlmExecution: true, hasLlmStreamExecution: true, + hasMarkSanitize: true, + hasScopeStartSanitize: true, + hasScopeEndSanitize: true, }); } finally { plugin.clear(); diff --git a/crates/node/tests/event_sanitizers_tests.mjs b/crates/node/tests/event_sanitizers_tests.mjs new file mode 100644 index 000000000..e4f776c9c --- /dev/null +++ b/crates/node/tests/event_sanitizers_tests.mjs @@ -0,0 +1,222 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import { createRequire } from 'node:module'; + +const require = createRequire(import.meta.url); +const lib = require('../index.js'); +const plugin = require('../plugin.js'); + +function capture(name) { + const events = []; + lib.registerSubscriber(name, (event) => events.push(event)); + return events; +} + +async function waitFor(events, count) { + for (let attempt = 0; attempt < 100 && events.length < count; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.ok(events.length >= count, `expected ${count} events, received ${events.length}`); +} + +describe('event sanitizer registries', () => { + it('orders mark sanitizers and supports field removal', async () => { + const events = capture('node-event-sanitize-order-sub'); + const calls = []; + lib.registerMarkSanitizeGuardrail('node-event-first', 10, (event, fields) => { + calls.push([event.name, fields.data]); + return { ...fields, data: { stage: 'first' }, metadata: null }; + }); + lib.registerMarkSanitizeGuardrail('node-event-second', 20, (event, fields) => { + calls.push([event.kind, fields.data]); + return { ...fields, data: { stage: 'second' } }; + }); + try { + lib.event('checkpoint', null, { secret: 'raw' }, { secret: 'raw' }); + lib.flushSubscribers(); + await waitFor(events, 1); + } finally { + lib.deregisterMarkSanitizeGuardrail('node-event-first'); + lib.deregisterMarkSanitizeGuardrail('node-event-second'); + lib.deregisterSubscriber('node-event-sanitize-order-sub'); + } + const mark = events.at(-1); + assert.deepEqual(mark.data, { stage: 'second' }); + assert.equal(mark.metadata, null); + assert.deepEqual(calls, [ + ['checkpoint', { secret: 'raw' }], + ['mark', { stage: 'first' }], + ]); + }); + + it('sanitizes scope start/end data, category profile, and metadata', async () => { + const events = capture('node-event-sanitize-scope-sub'); + const sanitize = (_event, fields) => ({ + data: null, + categoryProfile: { ...fields.categoryProfile, subtype: 'sanitized' }, + metadata: { safe: true }, + }); + lib.registerScopeSanitizeStartGuardrail('node-scope-start', 0, sanitize); + lib.registerScopeSanitizeEndGuardrail('node-scope-end', 0, sanitize); + try { + const handle = lib.pushScope( + 'generic', + lib.ScopeType.Custom, + null, + null, + { secret: 'start' }, + { secret: 'start' }, + { secret: 'input' }, + ); + lib.popScope(handle, { secret: 'output' }, null, { secret: 'end' }); + lib.flushSubscribers(); + await waitFor(events, 2); + } finally { + lib.deregisterScopeSanitizeStartGuardrail('node-scope-start'); + lib.deregisterScopeSanitizeEndGuardrail('node-scope-end'); + lib.deregisterSubscriber('node-event-sanitize-scope-sub'); + } + const lifecycle = events.filter((event) => event.name === 'generic'); + assert.equal(lifecycle.length, 2); + assert.ok(lifecycle.every((event) => event.data === null)); + assert.ok(lifecycle.every((event) => event.metadata.safe === true)); + assert.ok(lifecycle.every((event) => event.category_profile.subtype === 'sanitized')); + }); + + it('fails open and records invalid direct sanitizer results', async () => { + const events = capture('node-event-sanitize-invalid-sub'); + const invalidResults = { + scalar: () => 'invalid', + emptyObject: () => ({}), + array: () => [], + promise: () => Promise.resolve({ data: { changed: true } }), + }; + try { + for (const [kind, sanitizer] of Object.entries(invalidResults)) { + const name = `node-event-invalid-${kind}`; + lib.clearLastCallbackError(); + lib.registerMarkSanitizeGuardrail(name, 0, sanitizer); + try { + lib.event(name, null, { kept: kind }); + lib.flushSubscribers(); + await waitFor(events, Object.keys(invalidResults).indexOf(kind) + 1); + } finally { + lib.deregisterMarkSanitizeGuardrail(name); + } + assert.deepEqual(events.at(-1).data, { kept: kind }); + assert.match(lib.getLastCallbackError(), /event sanitizer callback failed/); + } + } finally { + lib.deregisterSubscriber('node-event-sanitize-invalid-sub'); + } + }); + + it('uses the thread-safe callback path for managed tool events', async () => { + const events = capture('node-event-sanitize-background-sub'); + lib.registerScopeSanitizeStartGuardrail('node-background-start', 0, (_event, fields) => ({ + ...fields, + metadata: { background: true }, + })); + try { + await lib.toolCallExecute('background-tool', { raw: true }, async (args) => args); + lib.flushSubscribers(); + await waitFor(events, 2); + } finally { + lib.deregisterScopeSanitizeStartGuardrail('node-background-start'); + lib.deregisterSubscriber('node-event-sanitize-background-sub'); + } + const start = events.find( + (event) => event.kind === 'scope' && event.name === 'background-tool' && event.scope_category === 'start', + ); + assert.equal(start.metadata.background, true); + }); + + it('fails open and records invalid thread-safe sanitizer results', async () => { + const events = capture('node-event-sanitize-background-invalid-sub'); + const invalidResults = { + emptyObject: () => ({}), + array: () => [], + promise: () => Promise.resolve({ data: { changed: true } }), + }; + try { + for (const [kind, sanitizer] of Object.entries(invalidResults)) { + const name = `node-background-invalid-${kind}`; + lib.clearLastCallbackError(); + lib.registerScopeSanitizeStartGuardrail(name, 0, sanitizer); + try { + await lib.toolCallExecute(name, { kept: kind }, async (args) => args); + lib.flushSubscribers(); + await waitFor(events, (Object.keys(invalidResults).indexOf(kind) + 1) * 2); + } finally { + lib.deregisterScopeSanitizeStartGuardrail(name); + } + const start = events.find( + (event) => event.kind === 'scope' && event.name === name && event.scope_category === 'start', + ); + assert.deepEqual(start.data, { kept: kind }); + assert.match(lib.getLastCallbackError(), /invalid JS event sanitizer result/); + } + } finally { + lib.deregisterSubscriber('node-event-sanitize-background-invalid-sub'); + } + }); + + it('inherits and cleans up scope-local mark sanitizers', async () => { + const events = capture('node-event-sanitize-local-sub'); + const owner = lib.pushScope('owner', lib.ScopeType.Agent); + lib.scopeRegisterMarkSanitizeGuardrail(owner.uuid, 'node-local-mark', 0, (_event, fields) => ({ + ...fields, + data: { local: true }, + })); + lib.event('inside', owner, { raw: true }); + const child = lib.pushScope('child', lib.ScopeType.Function, owner); + lib.event('inherited', child, { raw: true }); + lib.popScope(child); + lib.popScope(owner); + lib.event('outside', null, { raw: true }); + lib.flushSubscribers(); + await waitFor(events, 3); + lib.deregisterSubscriber('node-event-sanitize-local-sub'); + const marks = Object.fromEntries( + events.filter((event) => event.kind === 'mark').map((event) => [event.name, event]), + ); + assert.deepEqual(marks.inside.data, { local: true }); + assert.deepEqual(marks.inherited.data, { local: true }); + assert.deepEqual(marks.outside.data, { raw: true }); + }); + + it('cleans up plugin-owned event sanitizers', async () => { + const kind = `node.test.event-sanitize.${Date.now()}`; + const events = capture('node-event-sanitize-plugin-sub'); + plugin.register(kind, { + register(_config, context) { + context.registerMarkSanitizeGuardrail('mark', 0, (_event, fields) => ({ + ...fields, + data: { plugin: true }, + })); + }, + }); + try { + await plugin.initialize({ version: 1, components: [plugin.ComponentSpec(kind)] }); + lib.event('configured', null, { raw: true }); + lib.flushSubscribers(); + await waitFor(events, 1); + plugin.clear(); + lib.event('cleared', null, { raw: true }); + lib.flushSubscribers(); + await waitFor(events, 2); + } finally { + plugin.clear(); + plugin.deregister(kind); + lib.deregisterSubscriber('node-event-sanitize-plugin-sub'); + } + const marks = Object.fromEntries( + events.filter((event) => event.kind === 'mark').map((event) => [event.name, event]), + ); + assert.deepEqual(marks.configured.data, { plugin: true }); + assert.deepEqual(marks.cleared.data, { raw: true }); + }); +}); diff --git a/crates/node/tests/pii_redaction_tests.mjs b/crates/node/tests/pii_redaction_tests.mjs index dad3db91b..0c0edc3f7 100644 --- a/crates/node/tests/pii_redaction_tests.mjs +++ b/crates/node/tests/pii_redaction_tests.mjs @@ -18,6 +18,7 @@ describe('pii_redaction plugin helpers', () => { output: true, tool_input: true, tool_output: true, + mark: true, priority: 100, }); assert.deepEqual(piiRedaction.builtinConfig(), { action: 'remove' }); @@ -44,6 +45,9 @@ describe('pii_redaction plugin helpers', () => { }), ], }); - assert.deepEqual(report.diagnostics.map((diagnostic) => diagnostic.field), ['builtin.detector']); + assert.deepEqual( + report.diagnostics.map((diagnostic) => diagnostic.field), + ['builtin.detector'], + ); }); });