Build PluginsNative Dynamic Plugins

Observe and Sanitize

View as Markdown

The example’s observe feature group installs an Event metadata injector, subscriber, mark sanitizer, scope-start sanitizer, scope-end sanitizer, tool request and response sanitizers, and LLM request and response sanitizers. The group is enabled independently and receives the configured tag and redact_keys, so operators can see exactly which behavior one setting controls.

Register the Observation Surfaces

The function returns immediately when observation is disabled, so a report for that configuration cannot claim registrations that do not exist. One helper implements the shared event-field transformation, but each event surface is registered separately.

1pub(crate) fn register(
2    context: &mut PluginContext<'_>,
3    config: &ExampleConfig,
4    runtime: &PluginRuntime,
5) -> nemo_relay_plugin::Result<()> {
6    if !config.observe.enabled {
7        return Ok(());
8    }
9
10    context.register_subscriber("documentation_subscriber", {
11        let runtime = runtime.clone();
12        let tag = config.tag.clone();
13        move |event| subscriber_mark(&runtime, &tag, event)
14    })?;
15
16    let sanitize =
17        |fields: EventSanitizeFields, tag: String, keys: Vec<String>| async move {
18            sanitize_event_fields(fields, &tag, &keys)
19        };
20    context.register_mark_sanitize_guardrail("documentation_mark_sanitizer", 10, {
21        let tag = config.tag.clone();
22        let keys = config.observe.redact_keys.clone();
23        move |_event, fields| sanitize(fields, tag.clone(), keys.clone())
24    })?;
25    context.register_scope_sanitize_start_guardrail(
26        "documentation_scope_start_sanitizer",
27        10,
28        {
29            let tag = config.tag.clone();
30            let keys = config.observe.redact_keys.clone();
31            move |_event, fields| sanitize(fields, tag.clone(), keys.clone())
32        },
33    )?;
34    context.register_scope_sanitize_end_guardrail(
35        "documentation_scope_end_sanitizer",
36        10,
37        {
38            let tag = config.tag.clone();
39            let keys = config.observe.redact_keys.clone();
40            move |_event, fields| sanitize(fields, tag.clone(), keys.clone())
41        },
42    )?;
43    context.register_tool_sanitize_request_guardrail(
44        "documentation_tool_request_sanitizer",
45        10,
46        {
47            let keys = config.observe.redact_keys.clone();
48            move |_name, value| {
49                let keys = keys.clone();
50                async move { Ok(redact_json(value, &keys)) }
51            }
52        },
53    )?;
54    context.register_tool_sanitize_response_guardrail(
55        "documentation_tool_response_sanitizer",
56        10,
57        {
58            let keys = config.observe.redact_keys.clone();
59            move |_name, value| {
60                let keys = keys.clone();
61                async move { Ok(redact_json(value, &keys)) }
62            }
63        },
64    )?;
65    // The LLM sanitizer registrations are shown under "Use codecs for model payloads."
66    Ok(())
67}

The local name is unique only within this component. The closure clones owned strings and arrays into each async callback; it never borrows the component configuration after register returns.

Inject Event Metadata

Register a component-owned callback to propose metadata for every Event while the plugin is active. The callback receives an immutable Event snapshot and returns flat key/value additions:

1use std::collections::BTreeMap;
2
3use nemo_relay_plugin::Json;
4
5context.register_event_metadata_injector(
6    "documentation_event_metadata_injector",
7    5,
8    |_event| async move {
9        Ok(BTreeMap::from([(
10            "app.source".into(),
11            Json::String("plugin".into()),
12        )]))
13    },
14)?;

Relay validates and merges accepted additions before Event sanitizers run. A callback error omits that callback’s additions without dropping the Event. Clearing the plugin component removes the registration.

Keep Real Values Separate from Event Values

A sanitizer returns the fields Relay should publish. It never changes the request that the real callback receives or the response that the application receives. To prove that distinction, place a configured secret field in a managed tool request, observe the redacted start event, and confirm that the tool callback still receives the original value.

Event sanitizers receive an immutable event together with mutable copies of data, category_profile, and metadata. The example walks those JSON values recursively, replaces keys listed by observe.redact_keys, and adds the documentation tag to the returned metadata. It registers all three event-specific callbacks because mark, scope-start, and scope-end are separate public surfaces.

1fn sanitize_event_fields(
2    mut fields: EventSanitizeFields,
3    tag: &str,
4    redact_keys: &[String],
5) -> nemo_relay_plugin::Result<EventSanitizeFields> {
6    fields.data = fields.data.map(|value| redact_json(value, redact_keys));
7    fields.metadata = Some(tagged_metadata(fields.metadata, tag, redact_keys));
8    if let Some(profile) = fields.category_profile.take() {
9        let value = serde_json::to_value(profile)
10            .map_err(|error| error.to_string())?;
11        fields.category_profile = Some(
12            serde_json::from_value(redact_json(value, redact_keys))
13                .map_err(|error| error.to_string())?,
14        );
15    }
16    Ok(fields)
17}
18
19fn redact_json(value: Json, redact_keys: &[String]) -> Json {
20    match value {
21        Json::Object(mut object) => {
22            for (key, value) in &mut object {
23                if redact_keys.iter().any(|candidate| candidate == key) {
24                    *value = Json::String("[REDACTED]".into());
25                } else {
26                    *value = redact_json(value.take(), redact_keys);
27                }
28            }
29            Json::Object(object)
30        }
31        Json::Array(values) => Json::Array(
32            values.into_iter()
33                .map(|value| redact_json(value, redact_keys))
34                .collect(),
35        ),
36        other => other,
37    }
38}

The subscriber observes the sanitized event stream and emits no recursive event for an event already created by the example. Because native subscribers are synchronous, it does only bounded local work. Network export or other asynchronous I/O belongs in typed middleware or a purpose-built exporter with its own queue and shutdown behavior.

Use Codecs for Model Payloads

LLM sanitizers receive a request or response plus a structured context. For requests, the example resolves the directional codec when one is available, redacts the normalized annotation, and encodes it back onto the original envelope before applying its raw JSON fallback. Response codecs expose decode but not a symmetric response encoder, so the response sanitizer resolves and decodes the active codec for normalized inspection, then returns a redacted provider envelope. When the codec is absent or opaque, both callbacks redact the original JSON safely. Returning no payload omits that request or response from observability; it does not block the model call.

The codec facade is valid only for the callback lifetime. It can remain in the typed future across an await, because the SDK owns that lifetime, but it must not be cached in plugin-global state or used by later invocations.

1context.register_llm_sanitize_request_guardrail(
2    "documentation_llm_request_sanitizer",
3    10,
4    {
5        let redact_keys = config.observe.redact_keys.clone();
6        move |mut request, codec_context| {
7            let redact_keys = redact_keys.clone();
8            async move {
9                if let Some(codec) = codec_context.resolve_codec() {
10                    let annotated = codec.decode(&request)?;
11                    let annotated = serde_json::to_value(annotated)
12                        .map(|value| redact_json(value, &redact_keys))
13                        .and_then(serde_json::from_value)
14                        .map_err(|error| error.to_string())?;
15                    request = codec.encode(&annotated, &request)?;
16                }
17                request.content = redact_json(request.content, &redact_keys);
18                Ok(Some(request))
19            }
20        }
21    },
22)?;
23
24context.register_llm_sanitize_response_guardrail(
25    "documentation_llm_response_sanitizer",
26    10,
27    {
28        let redact_keys = config.observe.redact_keys.clone();
29        move |response, codec_context| {
30            let redact_keys = redact_keys.clone();
31            async move {
32                if let Some(codec) = codec_context.resolve_codec() {
33                    let _annotated = codec.decode(&response)?;
34                }
35                Ok(Some(redact_json(response, &redact_keys)))
36            }
37        }
38    },
39)?;

Verify Observation Behavior

Use the following procedure to verify all nine observation registrations without confusing observability changes with execution changes:

  1. Activate the example with observe.enabled = true, redact_keys = ["secret"], and the remaining feature groups disabled.
  2. Emit a mark and open and close a scope whose data, category profile, and metadata each contain a secret key. Capture the subscriber output.
  3. Execute a tool request and an LLM request and response containing the same key. Use a built-in codec for the model call and repeat once without a codec.
  4. Assert that every emitted observability field is redacted and tagged while the tool callback, model callback, and application result still contain their original real values.
  5. Clear the component and emit another mark. Confirm that neither sanitization nor the example subscriber runs.

Success means all nine observation registrations produce observable evidence, codec and fallback paths both redact safely, and no sanitizer accidentally changes execution.