feat(observability): route ATIF files with scope metadata

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<PathBuf>,
-    /// 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.<path>:-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(&section.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,21 +1372,22 @@ impl AtifDispatcher {
         }
     }
 
-    fn prepare_destination(&self, session_id: &str) -> (String, Option<PathBuf>) {
-        let filename = self
-            .config
-            .filename_template
-            .replace("{session_id}", session_id);
+    fn prepare_destination(
+        &self,
+        session_id: &str,
+        metadata: Option<&Json>,
+    ) -> Result<(String, Option<PathBuf>), 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
             .output_directory
             .clone()
             .unwrap_or_else(default_output_directory);
         let path = directory.join(&filename);
-        (filename, Some(path))
+        Ok((filename, Some(path)))
     }
 
     fn sink_targets(&self) -> Vec<SinkLabel> {
@@ -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<String, String> {
+    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<Mutex<AtifDispatcher>>,
     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(&section.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() {

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();