fix: harden trajectory context redaction

crates/pii-redaction/README.md

@@ -110,12 +110,11 @@ required. Profile-array mode covers marks, LLM and tool observability, and
 scope metadata automatically. The original single-policy surface flags remain
 available for backward compatibility but cannot be combined with `profiles`.
 
-### Structure-Preserving Trajectory Export
+### Fail-Closed Trajectory Export
 
 Use the `trajectory_context` preset when exported trajectories must retain
-their analytical structure without retaining chat, reasoning, tool, or
-multimodal content. Pair it with a later email profile so email addresses are
-also removed from otherwise-preserved metadata and custom marks:
+normalized analytical structure without retaining chat, reasoning, tool, or
+multimodal payloads:
 
 ```toml
 [[components]]
@@ -131,24 +130,29 @@ priority = 80
 
 [components.config.profiles.builtin]
 preset = "trajectory_context"
-custom_mark_payload_policy = "redact_all_leaves"
-
-[[components.config.profiles]]
-mode = "builtin"
-priority = 90
-
-[components.config.profiles.builtin]
-action = "redact"
-detector = "email"
 ```
 
-`custom_mark_payload_policy = "preserve"` is the default and leaves unknown
-plugin mark payloads intact for analysis. Use `redact_all_leaves` when opaque
-plugins may emit content: scalar leaves in data, metadata, and opaque category
-profile fields are replaced while typed category identity remains valid. Relay
-metric-schema marks use schema-aware sanitization instead: required measurement
-fields and numeric analytics remain valid for metric export, while descriptions
-and string attribute values are redacted.
+The preset empties raw LLM and tool payloads, headers, generic event data and
+metadata, provider-native values, category-profile extras, and unknown custom
+mark payloads. Present opaque JSON values become `{}`. The reserved
+`nemo_relay.log.severity` mark metadata field is restored in canonical form when
+valid. Set `custom_mark_payload_policy = "preserve"` only when an unknown custom
+mark producer is trusted and its original payload must remain available.
+
+Normalized LLM annotations retain message roles and content-part kinds; model,
+tool, provider, and metric names; request tuning parameters; finish reasons;
+token and cache usage; and normalized cost amounts, currency, and source
+classification. Content strings and application identifiers use the configured
+replacement, which defaults to `[REDACTED]`. Opaque typed fields become `{}`,
+unapproved numbers become `0`, and unapproved booleans become `false`.
+Pricing provenance, routing metadata, optimization contribution evidence, and
+unknown fields are discarded. Relay event, parent, trace, and span identifiers
+are not sanitizer fields, so trajectory hierarchy remains available.
+
+Provider, model, tool, and metric names are intentional free-form exceptions.
+Producers must not place user content, secrets, or personal identifiers in those
+name fields. Redacted application identifiers all use the same marker and must
+not be used as correlation keys.
 
 The preset can preserve explicitly approved, bounded string metric dimensions
 without exposing arbitrary text:
@@ -159,33 +163,13 @@ without exposing arbitrary text:
 ```
 
 Relay compares both the attribute name and value exactly and case-sensitively.
-For string arrays, each element is checked separately. Unlisted attributes and
-unexpected values remain redacted. The allowlist is empty by default and does
+For string arrays, every element must be allowed or Relay drops the whole
+attribute. Unlisted attributes, unexpected string values, and all numeric or
+boolean attributes are dropped. The allowlist is empty by default and does
 not apply to descriptions, mark metadata, category profiles, or non-metric
 marks. Configure only fixed constants or bounded enum values; do not add
 free-form identifiers or user-provided text.
 
-Strings become `[REDACTED]`, numbers become `0`, booleans become `false`, and
-nulls, keys, arrays, and object shape are retained. On every mark, the preset
-preserves the reserved `nemo_relay.log.severity` metadata field in canonical
-form when it contains a supported severity. For opaque custom marks that use
-`redact_all_leaves`, unsupported severity values remain redacted with the
-other string leaves.
-Known Relay marks are sanitized semantically so their structural and analytical
-fields remain usable. This choice affects canonical event fields before
-subscriber fan-out; exporter-owned resource attributes are outside this
-boundary.
-
-For Scope events, the preset retains direct string values for the trusted
-low-cardinality classification fields `nemo_relay_scope_role`, `agent_kind`,
-`hook_event_name`, `gateway_config_profile`, `gateway_mode`, `turn_source`,
-`harness`, `source`, `identity_quality`, `gateway_path`,
-`llm_correlation_status`, `llm_correlation_source`, `tool_correlation_status`,
-`tool_correlation_source`, `otel.status_code`, and `fidelity_source`. It also
-retains the direct boolean `provider_payload_exact`. Do not place PII or
-conversational content in these fields. Arbitrary metadata and unexpected value
-types continue through the preset's normal semantic redaction.
-
 The preset defines its own action and therefore cannot be combined with
 `action`, `detector`, `pattern`, `target_paths`, or mask-specific fields. Its
 optional `replacement` defaults to `[REDACTED]`.

crates/pii-redaction/src/builtin.rs

@@ -16,15 +16,18 @@ use nemo_relay::api::runtime::{
     BuiltinLlmCodec, EventSanitizeFn, LlmCodecIdentity, LlmSanitizeRequestFn,
     LlmSanitizeResponseFn, ToolSanitizeFn,
 };
-use nemo_relay::codec::request::AnnotatedLlmRequest;
+use nemo_relay::codec::request::{AnnotatedLlmRequest, Message, MessageContent};
 use nemo_relay::codec::resolve::{
     ProviderSurface, detect_response_surface, request_codec as build_request_codec,
     response_codec as build_response_codec,
 };
 use nemo_relay::codec::traits::{LlmCodec, LlmResponseCodec};
 use nemo_relay::plugin::{PluginError, Result as PluginResult};
 
-use super::component::{BuiltinBackendConfig, validate_metric_string_attribute_allowlist};
+use super::component::{
+    BuiltinBackendConfig, DEFAULT_CUSTOM_MARK_PAYLOAD_POLICY,
+    validate_metric_string_attribute_allowlist,
+};
 use super::detectors::BuiltinDetector;
 use super::overlay::BuiltinCodecName;
 use super::trajectory::{CustomMarkPayloadPolicy, TrajectorySanitizer, is_relay_metric_mark};
@@ -300,7 +303,9 @@ impl CompiledBuiltinBackend {
             }
             None => None,
         };
-        if trajectory.is_none() && config.custom_mark_payload_policy != "preserve" {
+        if trajectory.is_none()
+            && config.custom_mark_payload_policy != DEFAULT_CUSTOM_MARK_PAYLOAD_POLICY
+        {
             return Err(PluginError::InvalidConfig(
                 "builtin.custom_mark_payload_policy requires builtin.preset = 'trajectory_context'"
                     .to_string(),
@@ -638,6 +643,73 @@ impl CompiledBuiltinBackend {
             })
     }
 
+    fn sanitize_trajectory_request_with_codec(
+        &self,
+        trajectory: &TrajectorySanitizer,
+        codec: &dyn LlmCodec,
+        request: &LlmRequest,
+    ) -> Option<LlmRequest> {
+        let annotated = codec.decode(request).ok()?;
+        let sanitized = trajectory.sanitize_annotated_request(annotated)?;
+        if let Ok(request) = codec.encode(&sanitized, request) {
+            return Some(request);
+        }
+        self.encode_trajectory_request_items_incrementally(codec, request, &sanitized)
+    }
+
+    fn encode_trajectory_request_items_incrementally(
+        &self,
+        codec: &dyn LlmCodec,
+        request: &LlmRequest,
+        sanitized: &AnnotatedLlmRequest,
+    ) -> Option<LlmRequest> {
+        let mut output = request.clone();
+        for (index, message) in sanitized.messages.iter().enumerate() {
+            if let Some(MessageContent::Parts(parts)) = message_content(message) {
+                for (part_index, part) in parts.iter().enumerate() {
+                    let mut current = codec.decode(&output).ok()?;
+                    let current_message = current.messages.get_mut(index)?;
+                    let MessageContent::Parts(current_parts) =
+                        message_content_mut(current_message)?
+                    else {
+                        return None;
+                    };
+                    current_parts.get_mut(part_index)?.clone_from(part);
+                    output = codec.encode(&current, &output).ok()?;
+                }
+            }
+            if let Message::Assistant {
+                tool_calls: Some(tool_calls),
+                ..
+            } = message
+            {
+                for (call_index, call) in tool_calls.iter().enumerate() {
+                    let mut current = codec.decode(&output).ok()?;
+                    let Message::Assistant {
+                        tool_calls: Some(current_calls),
+                        ..
+                    } = current.messages.get_mut(index)?
+                    else {
+                        return None;
+                    };
+                    current_calls.get_mut(call_index)?.clone_from(call);
+                    output = codec.encode(&current, &output).ok()?;
+                }
+            }
+            let mut current = codec.decode(&output).ok()?;
+            current.messages[index] = message.clone();
+            output = codec.encode(&current, &output).ok()?;
+        }
+        if let Some(tools) = sanitized.tools.as_ref() {
+            for (index, tool) in tools.iter().enumerate() {
+                let mut current = codec.decode(&output).ok()?;
+                current.tools.as_mut()?.get_mut(index)?.clone_from(tool);
+                output = codec.encode(&current, &output).ok()?;
+            }
+        }
+        codec.encode(sanitized, &output).ok()
+    }
+
     fn sanitize_request_target_paths_incrementally(
         &self,
         codec: &dyn LlmCodec,
@@ -758,6 +830,21 @@ impl CompiledBuiltinBackend {
         .then_some(payload)
     }
 
+    fn sanitize_trajectory_response_with_codec(
+        &self,
+        trajectory: &TrajectorySanitizer,
+        codec: &dyn LlmResponseCodec,
+        surface: ProviderSurface,
+        payload: Json,
+    ) -> Option<Json> {
+        let annotated = codec.decode_response(&payload).ok()?;
+        let sanitized = trajectory.sanitize_annotated_response(annotated)?;
+        Some(
+            BuiltinCodecName::from_provider_surface(surface)
+                .overlay_response_payload(payload, &sanitized),
+        )
+    }
+
     fn normalized_response_targets_match(
         target_paths: &[Vec<String>],
         annotated: &Json,
@@ -780,6 +867,34 @@ impl CompiledBuiltinBackend {
     }
 }
 
+fn message_content(message: &Message) -> Option<&MessageContent> {
+    match message {
+        Message::System { content, .. }
+        | Message::User { content, .. }
+        | Message::Developer { content, .. }
+        | Message::Tool { content, .. } => Some(content),
+        Message::Assistant { content, .. } => content.as_ref(),
+        Message::Function { .. }
+        | Message::ToolCallItem { .. }
+        | Message::ToolResultItem { .. }
+        | Message::ProviderNative { .. } => None,
+    }
+}
+
+fn message_content_mut(message: &mut Message) -> Option<&mut MessageContent> {
+    match message {
+        Message::System { content, .. }
+        | Message::User { content, .. }
+        | Message::Developer { content, .. }
+        | Message::Tool { content, .. } => Some(content),
+        Message::Assistant { content, .. } => content.as_mut(),
+        Message::Function { .. }
+        | Message::ToolCallItem { .. }
+        | Message::ToolResultItem { .. }
+        | Message::ProviderNative { .. } => None,
+    }
+}
+
 pub(super) fn tool_sanitize_callback(backend: CompiledBuiltinBackend) -> ToolSanitizeFn {
     let backend = Arc::new(backend);
     Arc::new(move |_name: String, payload: Json| {
@@ -889,8 +1004,28 @@ pub(super) fn llm_sanitize_request_callback(
                     .as_object()
                     .cloned()
                     .unwrap_or_default();
-                request.content = trajectory.sanitize_provider_payload(request.content);
-                return Ok(Some(request));
+                let resolved = context.resolve_codec();
+                let fallback = if resolved.is_none() {
+                    backend
+                        .selected_surface(context.codec())
+                        .map(build_request_codec)
+                } else {
+                    None
+                };
+                let Some(codec) = resolved.as_deref().or(fallback.as_deref()) else {
+                    request.content = trajectory.sanitize_provider_payload(request.content);
+                    return Ok(Some(request));
+                };
+                let sanitized =
+                    backend.sanitize_trajectory_request_with_codec(trajectory, codec, &request);
+                if sanitized.is_none() {
+                    log_llm_payload_omitted(
+                        "request",
+                        context.codec(),
+                        "codec decode, typed sanitize, or encode failure",
+                    );
+                }
+                return Ok(sanitized);
             }
             request.headers = backend.sanitize_request_headers(request.headers);
             if backend.target_path_matcher.is_empty() {
@@ -930,7 +1065,28 @@ pub(super) fn llm_sanitize_response_callback(
         let backend = Arc::clone(&backend);
         Box::pin(async move {
             if let Some(trajectory) = backend.trajectory.as_ref() {
-                return Ok(Some(trajectory.sanitize_provider_payload(payload)));
+                let Some(surface) = backend.selected_surface(context.codec()) else {
+                    return Ok(Some(trajectory.sanitize_provider_payload(payload)));
+                };
+                let resolved = context.resolve_codec();
+                let fallback = if resolved.is_none() {
+                    Some(build_response_codec(surface))
+                } else {
+                    None
+                };
+                let Some(codec) = resolved.as_deref().or(fallback.as_deref()) else {
+                    return Ok(Some(trajectory.sanitize_provider_payload(payload)));
+                };
+                let sanitized = backend
+                    .sanitize_trajectory_response_with_codec(trajectory, codec, surface, payload);
+                if sanitized.is_none() {
+                    log_llm_payload_omitted(
+                        "response",
+                        context.codec(),
+                        "codec decode, typed sanitize, or encode failure",
+                    );
+                }
+                return Ok(sanitized);
             }
             if backend.target_path_matcher.is_empty() {
                 return Ok(Some(backend.sanitize_json_preorder_dfs(payload)));

crates/pii-redaction/src/component.rs

@@ -30,6 +30,7 @@ pub use super::local::{clear_local_backend_provider, register_local_backend_prov
 
 /// The plugin kind reserved for the built-in privacy component.
 pub const PII_REDACTION_PLUGIN_KIND: &str = "pii_redaction";
+pub(super) const DEFAULT_CUSTOM_MARK_PAYLOAD_POLICY: &str = "redact_all_leaves";
 
 /// Top-level PII redaction component wrapper.
 #[derive(Debug, Clone)]
@@ -509,7 +510,7 @@ fn custom_mark_payload_policy_schema(
     string_enum_schema(
         generator,
         &["preserve", "redact_all_leaves"],
-        Some("preserve"),
+        Some(DEFAULT_CUSTOM_MARK_PAYLOAD_POLICY),
     )
 }
 
@@ -1484,7 +1485,7 @@ fn default_builtin_action() -> String {
 }
 
 fn default_custom_mark_payload_policy() -> String {
-    "preserve".to_string()
+    DEFAULT_CUSTOM_MARK_PAYLOAD_POLICY.to_string()
 }
 
 fn default_true() -> bool {
@@ -1504,7 +1505,7 @@ fn is_default_builtin_action(action: &str) -> bool {
 }
 
 fn is_default_custom_mark_payload_policy(policy: &str) -> bool {
-    policy == "preserve"
+    policy == DEFAULT_CUSTOM_MARK_PAYLOAD_POLICY
 }
 
 fn is_true(value: &bool) -> bool {