diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index eebe80e36..e52f78cd3 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -256,8 +256,10 @@ pub struct AtifSectionConfig { /// [`storage`]: Self::storage #[serde(default, skip_serializing_if = "Option::is_none")] pub output_directory: Option, - /// Filename template. `{session_id}` is replaced with the top-level trajectory scope UUID. - /// When [`storage`] is non-empty, the rendered filename is appended to each backend's key prefix. + /// Filename template. `{session_id}` is replaced with the top-level trajectory scope UUID, and + /// `{metadata.:-fallback}` placeholders use path-safe strings from the top-level scope + /// metadata or the optional literal fallback. When [`storage`] is non-empty, the rendered + /// filename is appended to each backend's key prefix. /// /// [`storage`]: Self::storage #[serde(default = "default_atif_filename_template")] @@ -846,11 +848,8 @@ fn register_atif_dispatcher( section: AtifSectionConfig, ctx: &mut PluginRegistrationContext, ) -> PluginResult<()> { - if !section.filename_template.contains("{session_id}") { - return Err(PluginError::InvalidConfig( - "ATIF filename_template must contain '{session_id}'".to_string(), - )); - } + validate_atif_filename_template(§ion.filename_template) + .map_err(PluginError::InvalidConfig)?; let mut storage_vec = Vec::with_capacity(section.storage.len()); for (index, entry) in section.storage.iter().enumerate() { @@ -1171,9 +1170,22 @@ impl AtifDispatcher { // subscriber is attached after that start event has already been // emitted. let session_id = event.uuid().to_string(); + let (filename, local_path) = match self.prepare_destination(&session_id, event.metadata()) { + Ok(destination) => destination, + Err(error) => { + log::warn!( + target: "nemo_relay.observability", + event = "atif_destination_render_failed", + plugin_kind = OBSERVABILITY_PLUGIN_KIND, + exporter = "atif", + session_id = session_id.as_str(); + "ATIF destination rendering failed: {error}" + ); + return None; + } + }; let exporter = AtifExporter::new(session_id.clone(), self.agent_info()); (exporter.subscriber())(event); - let (filename, local_path) = self.prepare_destination(&session_id); let correlation = AtifCorrelation::from_event(event); self.scope_owners.insert(event.uuid(), event.uuid()); self.agents.insert( @@ -1360,13 +1372,14 @@ impl AtifDispatcher { } } - fn prepare_destination(&self, session_id: &str) -> (String, Option) { - let filename = self - .config - .filename_template - .replace("{session_id}", session_id); + fn prepare_destination( + &self, + session_id: &str, + metadata: Option<&Json>, + ) -> Result<(String, Option), String> { + let filename = render_atif_filename(&self.config.filename_template, session_id, metadata)?; if !self.config.storage.is_empty() { - return (filename, None); + return Ok((filename, None)); } let directory = self .config @@ -1374,7 +1387,7 @@ impl AtifDispatcher { .clone() .unwrap_or_else(default_output_directory); let path = directory.join(&filename); - (filename, Some(path)) + Ok((filename, Some(path))) } fn sink_targets(&self) -> Vec { @@ -1393,6 +1406,108 @@ impl AtifDispatcher { } } +fn is_valid_atif_metadata_selector(selector: &str) -> bool { + !selector.is_empty() + && selector.split('.').all(|segment| { + !segment.is_empty() + && segment + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) + }) +} + +fn parse_atif_metadata_expression(expression: &str) -> Result<(&str, Option<&str>), String> { + let (selector, fallback) = expression + .split_once(":-") + .map_or((expression, None), |(key, value)| (key, Some(value))); + if !is_valid_atif_metadata_selector(selector) { + return Err(format!( + "ATIF filename_template metadata placeholder '{{metadata.{selector}}}' must contain a dot-separated path of ASCII letters, digits, '-' or '_'" + )); + } + Ok((selector, fallback)) +} + +fn validate_atif_filename_template(template: &str) -> Result<(), String> { + const PREFIX: &str = "{metadata."; + + if !template.contains("{session_id}") { + return Err("ATIF filename_template must contain '{session_id}'".to_string()); + } + + let mut cursor = 0; + while let Some(relative_start) = template[cursor..].find(PREFIX) { + let selector_start = cursor + relative_start + PREFIX.len(); + let end = template[selector_start..] + .find('}') + .map(|relative_end| selector_start + relative_end) + .ok_or_else(|| { + "ATIF filename_template contains an unclosed metadata placeholder".to_string() + })?; + let (_, fallback) = parse_atif_metadata_expression(&template[selector_start..end])?; + if let Some(fallback) = fallback + && !is_safe_atif_metadata_path(fallback) + { + return Err(format!( + "ATIF filename_template fallback '{fallback}' must be a path-safe relative fragment" + )); + } + cursor = end + 1; + } + Ok(()) +} + +fn render_atif_filename( + template: &str, + session_id: &str, + metadata: Option<&Json>, +) -> Result { + const PREFIX: &str = "{metadata."; + + let mut rendered = template.replace("{session_id}", session_id); + let mut cursor = 0; + while let Some(relative_start) = rendered[cursor..].find(PREFIX) { + let start = cursor + relative_start; + let selector_start = start + PREFIX.len(); + let end = rendered[selector_start..] + .find('}') + .map(|relative_end| selector_start + relative_end) + .ok_or_else(|| { + "ATIF filename_template contains an unclosed metadata placeholder".to_string() + })?; + let expression = rendered[selector_start..end].to_string(); + let (selector, fallback) = parse_atif_metadata_expression(&expression)?; + let value = selector + .split('.') + .fold(metadata, |value, segment| value?.get(segment)) + .and_then(Json::as_str) + .or(fallback) + .ok_or_else(|| { + format!( + "filename_template placeholder '{{metadata.{selector}}}' must resolve to a string" + ) + })?; + if !is_safe_atif_metadata_path(value) { + return Err(format!( + "metadata path '{selector}' must be a path-safe relative fragment" + )); + } + rendered.replace_range(start..=end, value); + cursor = start + value.len(); + } + Ok(rendered) +} + +fn is_safe_atif_metadata_path(value: &str) -> bool { + !value.is_empty() + && value.split('/').all(|segment| { + !matches!(segment, "" | "." | "..") + && segment.bytes().all(|byte| { + byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') + }) + }) +} + fn atif_dispatcher_subscriber( manager: Arc>, subscriber_prefix: String, @@ -2376,14 +2491,14 @@ fn validate_atif_values( policy: &ConfigPolicy, section: &AtifSectionConfig, ) { - if !section.filename_template.contains("{session_id}") { + if let Err(message) = validate_atif_filename_template(§ion.filename_template) { push_policy_diag( diagnostics, policy.unsupported_value, "observability.unsupported_value", Some("atif".to_string()), Some("filename_template".to_string()), - "ATIF filename_template must contain '{session_id}'".to_string(), + message, ); } for (index, storage) in section.storage.iter().enumerate() { diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 128d473aa..6b17f869d 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -704,6 +704,35 @@ fn unknown_fields_and_bad_values_follow_policy() { assert!(ignore_report.diagnostics.is_empty()); } +#[test] +fn atif_filename_template_syntax_is_rejected_before_activation() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + + let valid_report = validate_plugin_config(&plugin_config(json!({ + "atif": { + "filename_template": "{metadata.workflow_id:-unassigned}/trajectory-{session_id}.json" + } + }))); + assert!(!valid_report.has_errors()); + + let malformed = "trajectory-{session_id}.json/{metadata.tenant"; + let invalid_report = validate_plugin_config(&plugin_config(json!({ + "atif": {"filename_template": malformed} + }))); + assert!(invalid_report.diagnostics.iter().any(|diag| { + diag.field.as_deref() == Some("filename_template") + && diag.message.contains("unclosed metadata placeholder") + })); + + let error = futures::executor::block_on(initialize_plugins_exact(plugin_config(json!({ + "policy": {"unsupported_value": "ignore"}, + "atif": {"enabled": true, "filename_template": malformed} + })))) + .unwrap_err(); + assert!(error.to_string().contains("unclosed metadata placeholder")); +} + #[test] fn invalid_shapes_and_strict_policy_are_reported() { let _guard = crate::observability::test_mutex().lock().unwrap(); @@ -1293,6 +1322,57 @@ fn atif_defaults_create_one_file_per_top_level_agent() { assert!(!second_serialized.contains("first-agent")); } +#[test] +fn atif_filename_template_routes_by_metadata_and_skips_invalid_paths() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-atif-metadata-template"); + + let config = plugin_config(json!({ + "atif": { + "enabled": true, + "output_directory": dir, + "filename_template": "{metadata.routing.artifact_path}/trajectory-{session_id}.json" + } + })); + futures::executor::block_on(initialize_plugins_exact(config)).unwrap(); + + let invalid = crate::api::scope::push_scope( + PushScopeParams::builder() + .name("invalid-metadata-path-agent") + .scope_type(ScopeType::Agent) + .metadata(json!({"routing": {"artifact_path": "../escape"}})) + .build(), + ) + .unwrap(); + pop(&invalid); + + let valid = crate::api::scope::push_scope( + PushScopeParams::builder() + .name("valid-metadata-path-agent") + .scope_type(ScopeType::Agent) + .metadata(json!({"routing": {"artifact_path": "tenant-a/session-123"}})) + .build(), + ) + .unwrap(); + pop(&valid); + + clear_plugin_configuration().unwrap(); + let invalid_filename = format!("trajectory-{}.json", invalid.uuid); + assert!( + !dir.join(&invalid_filename).exists() + && !dir.join("../escape").join(&invalid_filename).exists(), + "unsafe metadata path should not produce a trajectory file" + ); + assert!( + dir.join(format!( + "tenant-a/session-123/trajectory-{}.json", + valid.uuid + )) + .exists() + ); +} + #[test] fn atif_routes_global_descendant_events_by_parent_uuid() { let _guard = crate::observability::test_mutex().lock().unwrap(); @@ -1763,7 +1843,7 @@ fn write_atif_reports_missing_local_path_and_unregistered_remote_sink() { #[test] fn atif_dispatcher_default_output_path_uses_current_directory() { let dispatcher = AtifDispatcher::new(AtifSectionConfig::default()); - let (filename, local_path) = dispatcher.prepare_destination("session-1"); + let (filename, local_path) = dispatcher.prepare_destination("session-1", None).unwrap(); assert_eq!(filename, "nemo-relay-atif-session-1.json"); assert_eq!( local_path.unwrap(), @@ -1773,6 +1853,82 @@ fn atif_dispatcher_default_output_path_uses_current_directory() { ); } +#[test] +fn atif_metadata_template_values_must_be_safe_path_fragments() { + assert!( + validate_atif_filename_template( + "{metadata.workflow_id:-unassigned}/trajectory-{session_id}.json" + ) + .is_ok() + ); + assert_eq!( + render_atif_filename( + "{metadata.workflow_id:-unassigned}/trajectory-{session_id}.json", + "scope-id", + None + ) + .unwrap(), + "unassigned/trajectory-scope-id.json" + ); + + assert!(is_safe_atif_metadata_path( + "tenant-a/team_1.session-123~retry" + )); + + for value in [ + "", + "/absolute", + "trailing/", + "double//slash", + ".", + "../escape", + "tenant/../escape", + r"tenant\session", + "tenant name", + "tenant:session", + ] { + assert!( + !is_safe_atif_metadata_path(value), + "metadata path should be rejected: {value:?}" + ); + } + + let dispatcher = AtifDispatcher::new(AtifSectionConfig { + filename_template: "{metadata.artifact_path}/trajectory-{session_id}.json".to_string(), + ..AtifSectionConfig::default() + }); + assert!(dispatcher.prepare_destination("session-1", None).is_err()); + let non_string = json!({"artifact_path": 123}); + assert!( + dispatcher + .prepare_destination("session-1", Some(&non_string)) + .is_err() + ); + + for template in [ + "{metadata.}/trajectory-{session_id}.json", + "{metadata.tenant..id}/trajectory-{session_id}.json", + "{metadata.tenant/trajectory-{session_id}.json", + "{metadata.{tenant}}/trajectory-{session_id}.json", + "{metadata.missing:-../escape}/trajectory-{session_id}.json", + ] { + assert!( + validate_atif_filename_template(template).is_err(), + "template should fail configuration validation: {template:?}" + ); + let dispatcher = AtifDispatcher::new(AtifSectionConfig { + filename_template: template.to_string(), + ..AtifSectionConfig::default() + }); + assert!( + dispatcher + .prepare_destination("session-1", Some(&json!({"tenant": "tenant-a"}))) + .is_err(), + "template should be rejected: {template:?}" + ); + } +} + #[test] fn atif_payload_merges_correlation_with_existing_trajectory_extra() { let agent_uuid = Uuid::now_v7(); diff --git a/docs/configure-plugins/observability/atif.mdx b/docs/configure-plugins/observability/atif.mdx index 3acb8b227..c358404aa 100644 --- a/docs/configure-plugins/observability/atif.mdx +++ b/docs/configure-plugins/observability/atif.mdx @@ -68,9 +68,46 @@ The following table describes the top-level ATIF settings: | `tool_definitions` | Omitted | Optional ATIF tool metadata. | | `extra` | Omitted | Optional ATIF agent metadata. | | `output_directory` | Current working directory | Directory containing trajectory files. Ignored when `storage` is non-empty. | -| `filename_template` | `nemo-relay-atif-{session_id}.json` | Must contain `{session_id}`. | +| `filename_template` | `nemo-relay-atif-{session_id}.json` | Must contain `{session_id}`. Can contain `{metadata.}` placeholders for metadata-based routing. | | `storage` | Omitted | Optional list of remote storage destinations. When non-empty, trajectories are uploaded to every configured backend instead of being written locally. Refer to [Remote Storage](#remote-storage). | +### Metadata-Based Paths + +Use `{metadata.}` placeholders in `filename_template` to route +trajectories with top-level scope metadata. Dots select nested fields, and +templates can use multiple placeholders. For example: + +```toml +[components.config.atif] +enabled = true +output_directory = "logs" +filename_template = "{metadata.atif_prefix:-unassigned}/trajectory-{session_id}.json" +``` + +With scope metadata `{"atif_prefix":"tenant-a/session-123"}`, Relay writes +`logs/tenant-a/session-123/trajectory-.json`. The template must +still contain `{session_id}`. + +Use `:-` to provide a literal fallback when metadata can be absent, for example +`{metadata.atif_prefix:-unassigned}`. Relay also uses the fallback when the +metadata value is not a string. + +Each metadata placeholder must resolve to a string containing a non-empty, +relative path fragment. Slash-separated segments can contain ASCII letters, +digits, `-`, `_`, `.`, and `~`. Relay rejects empty segments, `.` and `..` +segments, absolute paths, backslashes, spaces, and other characters. If a +placeholder has no fallback and is missing or non-string, or if its resolved +value is unsafe, Relay skips that trajectory and logs +`atif_destination_render_failed`. The rendered filename applies to local, S3, +and HTTP storage in the same way as a static filename. + +The CLI gateway parses `x-nemo-relay-session-metadata` as JSON and merges it +into the top-level scope metadata: + +```http +x-nemo-relay-session-metadata: {"atif_prefix":"tenant-a/session-123"} +``` + ## Remote Storage Use `storage` when local trace files are not durable across sessions, such as in diff --git a/skills/nemo-relay-plugin-observability/references/atif.md b/skills/nemo-relay-plugin-observability/references/atif.md index febc22725..6e8f0226a 100644 --- a/skills/nemo-relay-plugin-observability/references/atif.md +++ b/skills/nemo-relay-plugin-observability/references/atif.md @@ -60,6 +60,16 @@ live OTLP spans. - Response codecs can improve LLM end annotations, but they do not change the caller-visible LLM response. +## Plugin-Managed File Routing + +- `filename_template` must contain `{session_id}` and can use + `{metadata.}` to select nested string values from top-level scope + metadata. Use `{metadata.:-fallback}` when the value is optional. +- Metadata values must be relative path fragments using ASCII letters, digits, + `-`, `_`, `.`, `~`, and safe `/`-separated segments. +- Missing, non-string, or unsafe metadata values skip only the affected + trajectory and produce an `atif_destination_render_failed` warning. + ## Checklist - [ ] Session and agent metadata chosen