diff --git a/crates/cli/tests/coverage/plugins_tests.rs b/crates/cli/tests/coverage/plugins_tests.rs
index fe36af1c7..17c03fd20 100644
--- a/crates/cli/tests/coverage/plugins_tests.rs
+++ b/crates/cli/tests/coverage/plugins_tests.rs
@@ -210,6 +210,13 @@ fn typed_editor_model_contains_observability_sections() {
             .iter()
             .any(|field| field.name == "endpoint")
     );
+    let attribute_mappings = openinference.field("attribute_mappings").unwrap();
+    assert_eq!(attribute_mappings.kind, EditorFieldKind::List);
+    let mapping = attribute_mappings.list_item.unwrap();
+    assert_eq!(mapping.kind, EditorFieldKind::Section);
+    let mapping_schema = mapping.schema.unwrap()();
+    assert_eq!(mapping_schema.fields[0].name, "key");
+    assert_eq!(mapping_schema.fields[1].name, "alias");
 }
 
 #[test]
diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs
index 6884cbd96..c3c6bf74c 100644
--- a/crates/core/src/observability/mod.rs
+++ b/crates/core/src/observability/mod.rs
@@ -6,6 +6,29 @@
 use crate::api::event::EventNormalizationExt;
 use serde::{Deserialize, Serialize};
 
+/// Copies a projected OTLP attribute to a second attribute name.
+///
+/// `key` names the fully-qualified projected attribute and `alias` names the
+/// additional attribute to emit with the same typed value.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
+pub struct OtlpAttributeMapping {
+    /// Fully-qualified projected attribute to copy.
+    pub key: String,
+    /// Additional attribute name receiving the copied value.
+    pub alias: String,
+}
+
+impl OtlpAttributeMapping {
+    /// Creates an attribute mapping.
+    pub fn new(key: impl Into<String>, alias: impl Into<String>) -> Self {
+        Self {
+            key: key.into(),
+            alias: alias.into(),
+        }
+    }
+}
+
 #[cfg(test)]
 use std::sync::Mutex;
 
@@ -49,6 +72,162 @@ pub(crate) fn default_mark_exclude_names() -> Vec<String> {
     vec!["llm.chunk".to_string()]
 }
 
+/// Validates OTLP attribute mappings shared by exporter configuration surfaces.
+pub fn validate_attribute_mappings(
+    mappings: &[OtlpAttributeMapping],
+) -> std::result::Result<(), String> {
+    let mut aliases = std::collections::HashSet::new();
+    for mapping in mappings {
+        if mapping.key.trim().is_empty() {
+            return Err("attribute mapping key must not be blank".to_string());
+        }
+        if mapping.alias.trim().is_empty() {
+            return Err("attribute mapping alias must not be blank".to_string());
+        }
+        if !aliases.insert(mapping.alias.trim()) {
+            return Err(format!(
+                "attribute mapping alias {:?} is duplicated",
+                mapping.alias
+            ));
+        }
+    }
+    Ok(())
+}
+
+#[cfg(any(feature = "otel", feature = "openinference"))]
+/// Projects only top-level JSON fields as OTLP attributes.
+///
+/// Nested objects and arrays remain JSON strings so arbitrary payloads do not
+/// create ambiguous dotted attribute paths or unbounded attribute sets.
+pub(crate) fn push_top_level_json_attributes(
+    attributes: &mut Vec<opentelemetry::KeyValue>,
+    prefix: &str,
+    value: Option<&crate::json::Json>,
+) {
+    let Some(value) = value else {
+        return;
+    };
+    match value {
+        crate::json::Json::Object(values) => {
+            for (field, value) in values {
+                push_top_level_json_value(attributes, &format!("{prefix}.{field}"), value);
+            }
+        }
+        value => push_top_level_json_value(attributes, prefix, value),
+    }
+}
+
+#[cfg(any(feature = "otel", feature = "openinference"))]
+/// Serializes a value and projects its top-level JSON fields as OTLP attributes.
+pub(crate) fn push_serialized_top_level_attributes<T: Serialize + ?Sized>(
+    attributes: &mut Vec<opentelemetry::KeyValue>,
+    prefix: &str,
+    value: Option<&T>,
+) {
+    let Some(value) = value else {
+        return;
+    };
+    if let Ok(value) = serde_json::to_value(value) {
+        push_top_level_json_attributes(attributes, prefix, Some(&value));
+    }
+}
+
+#[cfg(any(feature = "otel", feature = "openinference"))]
+fn push_top_level_json_value(
+    attributes: &mut Vec<opentelemetry::KeyValue>,
+    key: &str,
+    value: &crate::json::Json,
+) {
+    use opentelemetry::KeyValue;
+
+    match value {
+        crate::json::Json::Null => {}
+        crate::json::Json::Bool(value) => attributes.push(KeyValue::new(key.to_string(), *value)),
+        crate::json::Json::String(value) => {
+            attributes.push(KeyValue::new(key.to_string(), value.clone()))
+        }
+        crate::json::Json::Number(value) => {
+            if let Some(value) = value.as_i64() {
+                attributes.push(KeyValue::new(key.to_string(), value));
+            } else if let Some(value) = value.as_u64() {
+                if let Ok(value) = i64::try_from(value) {
+                    attributes.push(KeyValue::new(key.to_string(), value));
+                } else {
+                    attributes.push(KeyValue::new(key.to_string(), value.to_string()));
+                }
+            } else if let Some(value) = value.as_f64() {
+                attributes.push(KeyValue::new(key.to_string(), value));
+            }
+        }
+        crate::json::Json::Array(_) | crate::json::Json::Object(_) => {
+            if let Ok(value) = serde_json::to_string(value) {
+                attributes.push(KeyValue::new(key.to_string(), value));
+            }
+        }
+    }
+}
+
+#[cfg(any(feature = "otel", feature = "openinference"))]
+pub(crate) fn apply_attribute_mappings(
+    attributes: &mut Vec<opentelemetry::KeyValue>,
+    mappings: &[OtlpAttributeMapping],
+) {
+    attributes.extend(attribute_mapping_aliases(attributes, mappings));
+}
+
+/// Keeps the start attributes needed to resolve mappings at the end of a span.
+///
+/// The final span attributes must still take precedence over mapped aliases, so
+/// retain both mapped source keys and aliases that were already present at
+/// start. The span itself owns all other start attributes and does not need a
+/// second copy in the active-span state.
+#[cfg(any(feature = "otel", feature = "openinference"))]
+pub(crate) fn attribute_mapping_inputs(
+    attributes: &[opentelemetry::KeyValue],
+    mappings: &[OtlpAttributeMapping],
+) -> Vec<opentelemetry::KeyValue> {
+    attributes
+        .iter()
+        .filter(|attribute| {
+            mappings.iter().any(|mapping| {
+                attribute.key.as_str() == mapping.key || attribute.key.as_str() == mapping.alias
+            })
+        })
+        .cloned()
+        .collect()
+}
+
+/// Resolves typed aliases from a complete set of projected attributes.
+///
+/// Callers that project a span across multiple lifecycle events must pass every
+/// real span attribute so projected fields always take precedence over aliases.
+#[cfg(any(feature = "otel", feature = "openinference"))]
+pub(crate) fn attribute_mapping_aliases(
+    projected_attributes: &[opentelemetry::KeyValue],
+    mappings: &[OtlpAttributeMapping],
+) -> Vec<opentelemetry::KeyValue> {
+    if mappings.is_empty() {
+        return Vec::new();
+    }
+    let existing = projected_attributes
+        .iter()
+        .map(|attribute| attribute.key.as_str().to_string())
+        .collect::<std::collections::HashSet<_>>();
+    mappings
+        .iter()
+        .filter(|mapping| !existing.contains(mapping.alias.as_str()))
+        .filter_map(|mapping| {
+            projected_attributes
+                .iter()
+                .rev()
+                .find(|attribute| attribute.key.as_str() == mapping.key)
+                .map(|attribute| {
+                    opentelemetry::KeyValue::new(mapping.alias.clone(), attribute.value.clone())
+                })
+        })
+        .collect()
+}
+
 /// Returns whether a mark matches a configured projection exclusion.
 ///
 /// Agent hook adapters may preserve the canonical event name in metadata while
@@ -200,3 +379,7 @@ where
     };
     span.set_status(status);
 }
+
+#[cfg(all(test, any(feature = "otel", feature = "openinference")))]
+#[path = "../../tests/unit/observability/attribute_projection_tests.rs"]
+mod attribute_projection_tests;
diff --git a/crates/core/src/observability/openinference.rs b/crates/core/src/observability/openinference.rs
index 37aa8ea16..4e3d9c1a0 100644
--- a/crates/core/src/observability/openinference.rs
+++ b/crates/core/src/observability/openinference.rs
@@ -21,9 +21,11 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use super::{
-    MarkProjection, default_mark_exclude_names, effective_mark_projection,
+    MarkProjection, OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_aliases,
+    attribute_mapping_inputs, default_mark_exclude_names, effective_mark_projection,
     estimate_cost_for_response_or_model, estimate_cost_for_response_or_requested_model, manual,
-    merge_usage, model_name_for_llm_event,
+    merge_usage, model_name_for_llm_event, push_serialized_top_level_attributes,
+    push_top_level_json_attributes, validate_attribute_mappings,
 };
 use crate::api::event::{Event, EventNormalizationExt, ScopeCategory};
 use crate::api::runtime::EventSubscriberFn;
@@ -77,6 +79,9 @@ pub enum OpenInferenceError {
     /// The underlying tracer provider returned an error.
     #[error("OpenInference tracer provider error: {0}")]
     Provider(String),
+    /// Attribute mapping configuration was invalid.
+    #[error("invalid attribute mappings: {0}")]
+    InvalidAttributeMappings(String),
     /// Registration errors from the core runtime.
     #[error(transparent)]
     Core(#[from] FlowError),
@@ -104,6 +109,7 @@ pub struct OpenInferenceConfig {
     instrumentation_scope: String,
     mark_projection: MarkProjection,
     mark_exclude_names: Vec<String>,
+    attribute_mappings: Vec<OtlpAttributeMapping>,
     timeout: Duration,
     transport: OtlpTransport,
 }
@@ -120,6 +126,7 @@ impl Default for OpenInferenceConfig {
             instrumentation_scope: "nemo-relay-openinference".to_string(),
             mark_projection: MarkProjection::default(),
             mark_exclude_names: default_mark_exclude_names(),
+            attribute_mappings: Vec::new(),
             timeout: Duration::from_secs(3),
             transport: OtlpTransport::HttpBinary,
         }
@@ -207,6 +214,26 @@ impl OpenInferenceConfig {
         self.mark_exclude_names = names.into_iter().map(Into::into).collect();
         self
     }
+
+    /// Adds a typed attribute copy after event payload projection.
+    pub fn with_attribute_mapping(
+        mut self,
+        key: impl Into<String>,
+        alias: impl Into<String>,
+    ) -> Self {
+        self.attribute_mappings
+            .push(OtlpAttributeMapping::new(key, alias));
+        self
+    }
+
+    /// Replaces the configured typed attribute copies.
+    pub fn with_attribute_mappings<I>(mut self, mappings: I) -> Self
+    where
+        I: IntoIterator<Item = OtlpAttributeMapping>,
+    {
+        self.attribute_mappings = mappings.into_iter().collect();
+        self
+    }
 }
 
 /// OpenInference-backed NeMo Relay subscriber.
@@ -215,6 +242,27 @@ pub struct OpenInferenceSubscriber {
     inner: Arc<Inner>,
 }
 
+/// Options for constructing an OpenInference subscriber from an existing tracer provider.
+#[derive(Debug, Clone)]
+pub struct OpenInferenceSubscriberOptions {
+    /// How mark events are projected into the trace.
+    pub mark_projection: MarkProjection,
+    /// Mark names excluded from tool projection.
+    pub mark_exclude_names: Vec<String>,
+    /// Typed OTLP attributes copied to alias keys.
+    pub attribute_mappings: Vec<OtlpAttributeMapping>,
+}
+
+impl Default for OpenInferenceSubscriberOptions {
+    fn default() -> Self {
+        Self {
+            mark_projection: MarkProjection::default(),
+            mark_exclude_names: default_mark_exclude_names(),
+            attribute_mappings: Vec::new(),
+        }
+    }
+}
+
 struct Inner {
     processor: Arc<Mutex<OpenInferenceEventProcessor>>,
     subscriber: EventSubscriberFn,
@@ -227,6 +275,8 @@ impl OpenInferenceSubscriber {
         {
             return Err(OpenInferenceError::MissingTokioRuntime);
         }
+        validate_attribute_mappings(&config.attribute_mappings)
+            .map_err(OpenInferenceError::InvalidAttributeMappings)?;
 
         let provider = build_tracer_provider(&config)?;
         Ok(Self::from_tracer_provider_with_scope(
@@ -234,6 +284,7 @@ impl OpenInferenceSubscriber {
             config.instrumentation_scope,
             config.mark_projection,
             config.mark_exclude_names,
+            config.attribute_mappings,
         ))
     }
 
@@ -247,6 +298,7 @@ impl OpenInferenceSubscriber {
             instrumentation_scope.into(),
             MarkProjection::default(),
             default_mark_exclude_names(),
+            Vec::new(),
         )
     }
 
@@ -261,6 +313,7 @@ impl OpenInferenceSubscriber {
             instrumentation_scope.into(),
             mark_projection,
             default_mark_exclude_names(),
+            Vec::new(),
         )
     }
 
@@ -280,21 +333,61 @@ impl OpenInferenceSubscriber {
             instrumentation_scope.into(),
             mark_projection,
             mark_exclude_names.into_iter().map(Into::into).collect(),
+            Vec::new(),
         )
     }
 
+    /// Builds a subscriber from a tracer provider with typed attribute copies.
+    pub fn from_tracer_provider_with_attribute_mappings<I>(
+        provider: SdkTracerProvider,
+        instrumentation_scope: impl Into<String>,
+        attribute_mappings: I,
+    ) -> Result<Self>
+    where
+        I: IntoIterator<Item = OtlpAttributeMapping>,
+    {
+        let attribute_mappings = attribute_mappings.into_iter().collect::<Vec<_>>();
+        Self::from_tracer_provider_with_options(
+            provider,
+            instrumentation_scope,
+            OpenInferenceSubscriberOptions {
+                attribute_mappings,
+                ..Default::default()
+            },
+        )
+    }
+
+    /// Builds a subscriber from a tracer provider with composable projection options.
+    pub fn from_tracer_provider_with_options(
+        provider: SdkTracerProvider,
+        instrumentation_scope: impl Into<String>,
+        options: OpenInferenceSubscriberOptions,
+    ) -> Result<Self> {
+        validate_attribute_mappings(&options.attribute_mappings)
+            .map_err(OpenInferenceError::InvalidAttributeMappings)?;
+        Ok(Self::from_tracer_provider_with_scope(
+            provider,
+            instrumentation_scope.into(),
+            options.mark_projection,
+            options.mark_exclude_names,
+            options.attribute_mappings,
+        ))
+    }
+
     fn from_tracer_provider_with_scope(
         provider: SdkTracerProvider,
         instrumentation_scope: String,
         mark_projection: MarkProjection,
         mark_exclude_names: Vec<String>,
+        attribute_mappings: Vec<OtlpAttributeMapping>,
     ) -> Self {
         let processor = Arc::new(Mutex::new(
-            OpenInferenceEventProcessor::new_with_mark_projection_and_exclusions(
+            OpenInferenceEventProcessor::new_with_mark_projection_and_exclusions_and_mappings(
                 provider,
                 instrumentation_scope,
                 mark_projection,
                 mark_exclude_names,
+                attribute_mappings,
             ),
         ));
         let processor_for_callback = Arc::clone(&processor);
@@ -442,6 +535,7 @@ fn build_grpc_metadata(headers: &HashMap<String, String>) -> Result<MetadataMap>
 struct ActiveSpan {
     span: Span,
     span_context: SpanContext,
+    projected_attributes: Vec<KeyValue>,
 }
 
 struct OpenInferenceEventProcessor {
@@ -452,6 +546,7 @@ struct OpenInferenceEventProcessor {
     tracer: SdkTracer,
     mark_projection: MarkProjection,
     mark_exclude_names: Vec<String>,
+    attribute_mappings: Vec<OtlpAttributeMapping>,
 }
 
 impl OpenInferenceEventProcessor {
@@ -474,11 +569,28 @@ impl OpenInferenceEventProcessor {
         )
     }
 
+    #[cfg(test)]
     fn new_with_mark_projection_and_exclusions(
         provider: SdkTracerProvider,
         instrumentation_scope: String,
         mark_projection: MarkProjection,
         mark_exclude_names: Vec<String>,
+    ) -> Self {
+        Self::new_with_mark_projection_and_exclusions_and_mappings(
+            provider,
+            instrumentation_scope,
+            mark_projection,
+            mark_exclude_names,
+            Vec::new(),
+        )
+    }
+
+    fn new_with_mark_projection_and_exclusions_and_mappings(
+        provider: SdkTracerProvider,
+        instrumentation_scope: String,
+        mark_projection: MarkProjection,
+        mark_exclude_names: Vec<String>,
+        attribute_mappings: Vec<OtlpAttributeMapping>,
     ) -> Self {
         let tracer = provider.tracer(instrumentation_scope);
         Self {
@@ -489,6 +601,7 @@ impl OpenInferenceEventProcessor {
             tracer,
             mark_projection,
             mark_exclude_names,
+            attribute_mappings,
         }
     }
 
@@ -520,10 +633,18 @@ impl OpenInferenceEventProcessor {
             .with_kind(span_kind(event))
             .with_start_time(to_system_time(*event.timestamp()))
             .start_with_context(&self.tracer, &self.parent_context(event));
-        span.set_attributes(start_attributes(event));
+        let attributes = start_attributes(event);
+        let projected_attributes = attribute_mapping_inputs(&attributes, &self.attribute_mappings);
+        span.set_attributes(attributes);
         let span_context = local_parent_span_context(span.span_context());
-        self.active_spans
-            .insert(event.uuid(), ActiveSpan { span, span_context });
+        self.active_spans.insert(
+            event.uuid(),
+            ActiveSpan {
+                span,
+                span_context,
+                projected_attributes,
+            },
+        );
     }
 
     fn process_end(&mut self, event: &Event) {
@@ -532,7 +653,16 @@ impl OpenInferenceEventProcessor {
         };
         self.record_completed_span_context(event.uuid(), active_span.span_context.clone());
         super::set_span_status_from_event_metadata(&mut active_span.span, event);
-        active_span.span.set_attributes(end_attributes(event));
+        let mut attributes = end_attributes(event);
+        if !self.attribute_mappings.is_empty() {
+            let mut projected_attributes = active_span.projected_attributes;
+            projected_attributes.extend(attributes.iter().cloned());
+            attributes.extend(attribute_mapping_aliases(
+                &projected_attributes,
+                &self.attribute_mappings,
+            ));
+        }
+        active_span.span.set_attributes(attributes);
         active_span
             .span
             .end_with_timestamp(to_system_time(*event.timestamp()));
@@ -547,9 +677,13 @@ impl OpenInferenceEventProcessor {
         }
         let mark_name = event.name().to_string();
         let timestamp = to_system_time(*event.timestamp());
-        let attributes = mark_attributes(event);
+        let mut attributes = mark_attributes(event);
 
-        if let Some(parent_span) = self.find_parent_span_mut(event) {
+        if self.find_parent_span(event).is_some() {
+            apply_attribute_mappings(&mut attributes, &self.attribute_mappings);
+            let parent_span = self
+                .find_parent_span_mut(event)
+                .expect("parent span was present during mark projection");
             parent_span
                 .span
                 .add_event_with_timestamp(mark_name, timestamp, attributes);
@@ -562,13 +696,13 @@ impl OpenInferenceEventProcessor {
             .with_kind(SpanKind::Internal)
             .with_start_time(timestamp)
             .start_with_context(&self.tracer, &self.parent_context(event));
-        let mut span_attributes = attributes;
-        span_attributes.push(KeyValue::new(
+        attributes.push(KeyValue::new(
             oi::OPENINFERENCE_SPAN_KIND,
             OpenInferenceSpanKind::Chain,
         ));
-        span_attributes.push(KeyValue::new("nemo_relay.mark.orphan", true));
-        span.set_attributes(span_attributes);
+        attributes.push(KeyValue::new("nemo_relay.mark.orphan", true));
+        apply_attribute_mappings(&mut attributes, &self.attribute_mappings);
+        span.set_attributes(attributes);
         span.end_with_timestamp(timestamp);
     }
 
@@ -584,6 +718,7 @@ impl OpenInferenceEventProcessor {
         if orphan {
             attributes.push(KeyValue::new("nemo_relay.mark.orphan", true));
         }
+        apply_attribute_mappings(&mut attributes, &self.attribute_mappings);
 
         let mut span = self
             .tracer
@@ -682,29 +817,24 @@ fn start_attributes(event: &Event) -> Vec<KeyValue> {
     let mut attributes = common_attributes(event);
     let is_llm = event
         .category()
-        .is_some_and(|category| category.as_str() == "llm");
+        .is_some_and(|category| category.as_str() == "llm")
+        || semantic_scope_type(event) == Some(ScopeType::Llm);
     if is_llm {
         // Final span metadata should reflect the completed event, especially for mixed-fidelity
         // Hermes flows where the request can be exact but the terminal error is lossy.
-        attributes.retain(|attribute| attribute.key.as_str() != oi::METADATA.as_str());
-    }
-    let handle_attributes = event.attributes();
-    if handle_attributes.is_some_and(|attributes| !attributes.is_empty()) {
-        push_serialized(
-            &mut attributes,
-            "nemo_relay.handle_attributes_json",
-            handle_attributes,
-        );
+        attributes.retain(|attribute| {
+            attribute.key.as_str() != oi::METADATA.as_str()
+                && !attribute.key.as_str().starts_with("openinference.metadata")
+        });
     }
-    if event
-        .category()
-        .is_none_or(|category| category.as_str() != "llm")
-    {
-        push_serialized(
+    if !is_llm {
+        push_serialized_top_level_attributes(
             &mut attributes,
-            "nemo_relay.start.input_json",
-            event.input(),
+            "nemo_relay.handle_attributes",
+            event.attributes(),
         );
+        push_top_level_json_attributes(&mut attributes, "nemo_relay.start.data", event.data());
+        push_top_level_json_attributes(&mut attributes, "nemo_relay.start.input", event.input());
     }
     if event
         .category()
@@ -739,17 +869,15 @@ fn end_attributes(event: &Event) -> Vec<KeyValue> {
     let mut attributes = Vec::new();
     let is_llm = event
         .category()
-        .is_some_and(|category| category.as_str() == "llm");
+        .is_some_and(|category| category.as_str() == "llm")
+        || semantic_scope_type(event) == Some(ScopeType::Llm);
 
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.end.data", event.data());
     if let Some(metadata) = event.metadata().and_then(to_json_string) {
         attributes.push(KeyValue::new(oi::METADATA, metadata));
     }
-
-    push_serialized(
-        &mut attributes,
-        "nemo_relay.end.output_json",
-        event.output(),
-    );
+    push_top_level_json_attributes(&mut attributes, "openinference.metadata", event.metadata());
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.end.output", event.output());
     if let Some((output, mime_type)) = openinference_output_value(event) {
         attributes.push(KeyValue::new(oi::output::VALUE, output));
         attributes.push(KeyValue::new(oi::output::MIME_TYPE, mime_type));
@@ -1357,7 +1485,6 @@ fn cost_total_from_llm_event(
 }
 
 fn mark_attributes(event: &Event) -> Vec<KeyValue> {
-    let handle_attributes = event.attributes();
     let mut attributes = vec![
         KeyValue::new("nemo_relay.mark.uuid", event.uuid().to_string()),
         KeyValue::new(
@@ -1368,15 +1495,15 @@ fn mark_attributes(event: &Event) -> Vec<KeyValue> {
                 .unwrap_or_default(),
         ),
     ];
-    push_serialized(
+    push_serialized_top_level_attributes(
         &mut attributes,
-        "nemo_relay.mark.attributes_json",
-        handle_attributes,
+        "nemo_relay.mark.attributes",
+        event.attributes(),
     );
-    push_serialized(&mut attributes, "nemo_relay.mark.data_json", event.data());
-    push_serialized(
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.mark.data", event.data());
+    push_top_level_json_attributes(
         &mut attributes,
-        "nemo_relay.mark.metadata_json",
+        "nemo_relay.mark.metadata",
         event.metadata(),
     );
     if let Some(category) = event.category() {
@@ -1385,9 +1512,9 @@ fn mark_attributes(event: &Event) -> Vec<KeyValue> {
             category.as_str().to_string(),
         ));
     }
-    push_serialized(
+    push_serialized_top_level_attributes(
         &mut attributes,
-        "nemo_relay.mark.category_profile_json",
+        "nemo_relay.mark.category_profile",
         event.category_profile(),
     );
     attributes
@@ -1422,6 +1549,7 @@ fn common_attributes(event: &Event) -> Vec<KeyValue> {
     if let Some(metadata) = event.metadata().and_then(to_json_string) {
         attributes.push(KeyValue::new(oi::METADATA, metadata));
     }
+    push_top_level_json_attributes(&mut attributes, "openinference.metadata", event.metadata());
 
     attributes
 }
@@ -1442,18 +1570,6 @@ fn openinference_span_kind(scope_type: Option<ScopeType>) -> OpenInferenceSpanKi
     }
 }
 
-fn push_serialized<T: Serialize + ?Sized>(
-    attributes: &mut Vec<KeyValue>,
-    key: &'static str,
-    value: Option<&T>,
-) {
-    if let Some(value) = value
-        && let Ok(json) = serde_json::to_string(value)
-    {
-        attributes.push(KeyValue::new(key, json));
-    }
-}
-
 fn openinference_input_value(event: &Event) -> Option<(String, &'static str)> {
     let input = event.input()?;
 
diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs
index 9624b0527..671367036 100644
--- a/crates/core/src/observability/otel.rs
+++ b/crates/core/src/observability/otel.rs
@@ -21,9 +21,11 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use super::{
-    MarkProjection, default_mark_exclude_names, effective_mark_projection,
+    MarkProjection, OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_aliases,
+    attribute_mapping_inputs, default_mark_exclude_names, effective_mark_projection,
     estimate_cost_for_response_or_model, estimate_cost_for_response_or_requested_model, manual,
-    model_name_for_llm_event,
+    model_name_for_llm_event, push_serialized_top_level_attributes, push_top_level_json_attributes,
+    validate_attribute_mappings,
 };
 use crate::api::event::{Event, EventNormalizationExt, ScopeCategory};
 use crate::api::runtime::EventSubscriberFn;
@@ -39,7 +41,6 @@ use opentelemetry::{Context, KeyValue};
 use opentelemetry_otlp::{Protocol, SpanExporter, WithExportConfig, WithHttpConfig};
 use opentelemetry_sdk::Resource;
 use opentelemetry_sdk::trace::{SdkTracer, SdkTracerProvider, Span};
-use serde::Serialize;
 use uuid::Uuid;
 
 const COMPLETED_SPAN_CONTEXT_LIMIT: usize = 4096;
@@ -71,6 +72,9 @@ pub enum OpenTelemetryError {
     /// The underlying tracer provider returned an error.
     #[error("OpenTelemetry tracer provider error: {0}")]
     Provider(String),
+    /// Attribute mapping configuration was invalid.
+    #[error("invalid attribute mappings: {0}")]
+    InvalidAttributeMappings(String),
     /// Registration errors from the core runtime.
     #[error(transparent)]
     Core(#[from] FlowError),
@@ -98,6 +102,7 @@ pub struct OpenTelemetryConfig {
     instrumentation_scope: String,
     mark_projection: MarkProjection,
     mark_exclude_names: Vec<String>,
+    attribute_mappings: Vec<OtlpAttributeMapping>,
     timeout: Duration,
     transport: OtlpTransport,
 }
@@ -114,6 +119,7 @@ impl Default for OpenTelemetryConfig {
             instrumentation_scope: "nemo-relay-otel".to_string(),
             mark_projection: MarkProjection::default(),
             mark_exclude_names: default_mark_exclude_names(),
+            attribute_mappings: Vec::new(),
             timeout: Duration::from_secs(3),
             transport: OtlpTransport::HttpBinary,
         }
@@ -202,6 +208,26 @@ impl OpenTelemetryConfig {
         self.mark_exclude_names = names.into_iter().map(Into::into).collect();
         self
     }
+
+    /// Adds a typed attribute copy after event payload projection.
+    pub fn with_attribute_mapping(
+        mut self,
+        key: impl Into<String>,
+        alias: impl Into<String>,
+    ) -> Self {
+        self.attribute_mappings
+            .push(OtlpAttributeMapping::new(key, alias));
+        self
+    }
+
+    /// Replaces the configured typed attribute copies.
+    pub fn with_attribute_mappings<I>(mut self, mappings: I) -> Self
+    where
+        I: IntoIterator<Item = OtlpAttributeMapping>,
+    {
+        self.attribute_mappings = mappings.into_iter().collect();
+        self
+    }
 }
 
 /// OpenTelemetry-backed NeMo Relay subscriber.
@@ -210,6 +236,27 @@ pub struct OpenTelemetrySubscriber {
     inner: Arc<Inner>,
 }
 
+/// Options for constructing an OpenTelemetry subscriber from an existing tracer provider.
+#[derive(Debug, Clone)]
+pub struct OpenTelemetrySubscriberOptions {
+    /// How mark events are projected into the trace.
+    pub mark_projection: MarkProjection,
+    /// Mark names excluded from tool projection.
+    pub mark_exclude_names: Vec<String>,
+    /// Typed OTLP attributes copied to alias keys.
+    pub attribute_mappings: Vec<OtlpAttributeMapping>,
+}
+
+impl Default for OpenTelemetrySubscriberOptions {
+    fn default() -> Self {
+        Self {
+            mark_projection: MarkProjection::default(),
+            mark_exclude_names: default_mark_exclude_names(),
+            attribute_mappings: Vec::new(),
+        }
+    }
+}
+
 struct Inner {
     processor: Arc<Mutex<OtelEventProcessor>>,
     subscriber: EventSubscriberFn,
@@ -222,6 +269,8 @@ impl OpenTelemetrySubscriber {
         {
             return Err(OpenTelemetryError::MissingTokioRuntime);
         }
+        validate_attribute_mappings(&config.attribute_mappings)
+            .map_err(OpenTelemetryError::InvalidAttributeMappings)?;
 
         let provider = build_tracer_provider(&config)?;
         Ok(Self::from_tracer_provider_with_scope(
@@ -229,6 +278,7 @@ impl OpenTelemetrySubscriber {
             config.instrumentation_scope,
             config.mark_projection,
             config.mark_exclude_names,
+            config.attribute_mappings,
         ))
     }
 
@@ -242,6 +292,7 @@ impl OpenTelemetrySubscriber {
             instrumentation_scope.into(),
             MarkProjection::default(),
             default_mark_exclude_names(),
+            Vec::new(),
         )
     }
 
@@ -256,6 +307,7 @@ impl OpenTelemetrySubscriber {
             instrumentation_scope.into(),
             mark_projection,
             default_mark_exclude_names(),
+            Vec::new(),
         )
     }
 
@@ -275,21 +327,61 @@ impl OpenTelemetrySubscriber {
             instrumentation_scope.into(),
             mark_projection,
             mark_exclude_names.into_iter().map(Into::into).collect(),
+            Vec::new(),
         )
     }
 
+    /// Builds a subscriber from a tracer provider with typed attribute copies.
+    pub fn from_tracer_provider_with_attribute_mappings<I>(
+        provider: SdkTracerProvider,
+        instrumentation_scope: impl Into<String>,
+        attribute_mappings: I,
+    ) -> Result<Self>
+    where
+        I: IntoIterator<Item = OtlpAttributeMapping>,
+    {
+        let attribute_mappings = attribute_mappings.into_iter().collect::<Vec<_>>();
+        Self::from_tracer_provider_with_options(
+            provider,
+            instrumentation_scope,
+            OpenTelemetrySubscriberOptions {
+                attribute_mappings,
+                ..Default::default()
+            },
+        )
+    }
+
+    /// Builds a subscriber from a tracer provider with composable projection options.
+    pub fn from_tracer_provider_with_options(
+        provider: SdkTracerProvider,
+        instrumentation_scope: impl Into<String>,
+        options: OpenTelemetrySubscriberOptions,
+    ) -> Result<Self> {
+        validate_attribute_mappings(&options.attribute_mappings)
+            .map_err(OpenTelemetryError::InvalidAttributeMappings)?;
+        Ok(Self::from_tracer_provider_with_scope(
+            provider,
+            instrumentation_scope.into(),
+            options.mark_projection,
+            options.mark_exclude_names,
+            options.attribute_mappings,
+        ))
+    }
+
     fn from_tracer_provider_with_scope(
         provider: SdkTracerProvider,
         instrumentation_scope: String,
         mark_projection: MarkProjection,
         mark_exclude_names: Vec<String>,
+        attribute_mappings: Vec<OtlpAttributeMapping>,
     ) -> Self {
         let processor = Arc::new(Mutex::new(
-            OtelEventProcessor::new_with_mark_projection_and_exclusions(
+            OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings(
                 provider,
                 instrumentation_scope,
                 mark_projection,
                 mark_exclude_names,
+                attribute_mappings,
             ),
         ));
         let processor_for_callback = Arc::clone(&processor);
@@ -436,6 +528,7 @@ fn build_grpc_metadata(headers: &HashMap<String, String>) -> Result<MetadataMap>
 struct ActiveSpan {
     span: Span,
     span_context: SpanContext,
+    projected_attributes: Vec<KeyValue>,
 }
 
 struct OtelEventProcessor {
@@ -446,6 +539,7 @@ struct OtelEventProcessor {
     tracer: SdkTracer,
     mark_projection: MarkProjection,
     mark_exclude_names: Vec<String>,
+    attribute_mappings: Vec<OtlpAttributeMapping>,
 }
 
 impl OtelEventProcessor {
@@ -468,11 +562,28 @@ impl OtelEventProcessor {
         )
     }
 
+    #[cfg(test)]
     fn new_with_mark_projection_and_exclusions(
         provider: SdkTracerProvider,
         instrumentation_scope: String,
         mark_projection: MarkProjection,
         mark_exclude_names: Vec<String>,
+    ) -> Self {
+        Self::new_with_mark_projection_and_exclusions_and_mappings(
+            provider,
+            instrumentation_scope,
+            mark_projection,
+            mark_exclude_names,
+            Vec::new(),
+        )
+    }
+
+    fn new_with_mark_projection_and_exclusions_and_mappings(
+        provider: SdkTracerProvider,
+        instrumentation_scope: String,
+        mark_projection: MarkProjection,
+        mark_exclude_names: Vec<String>,
+        attribute_mappings: Vec<OtlpAttributeMapping>,
     ) -> Self {
         let tracer = provider.tracer(instrumentation_scope);
         Self {
@@ -483,6 +594,7 @@ impl OtelEventProcessor {
             tracer,
             mark_projection,
             mark_exclude_names,
+            attribute_mappings,
         }
     }
 
@@ -514,10 +626,18 @@ impl OtelEventProcessor {
             .with_kind(span_kind(event))
             .with_start_time(to_system_time(*event.timestamp()))
             .start_with_context(&self.tracer, &self.parent_context(event));
-        span.set_attributes(start_attributes(event));
+        let attributes = start_attributes(event);
+        let projected_attributes = attribute_mapping_inputs(&attributes, &self.attribute_mappings);
+        span.set_attributes(attributes);
         let span_context = local_parent_span_context(span.span_context());
-        self.active_spans
-            .insert(event.uuid(), ActiveSpan { span, span_context });
+        self.active_spans.insert(
+            event.uuid(),
+            ActiveSpan {
+                span,
+                span_context,
+                projected_attributes,
+            },
+        );
     }
 
     fn process_end(&mut self, event: &Event) {
@@ -527,7 +647,16 @@ impl OtelEventProcessor {
         self.record_completed_span_context(event.uuid(), active_span.span_context.clone());
 
         super::set_span_status_from_event_metadata(&mut active_span.span, event);
-        active_span.span.set_attributes(end_attributes(event));
+        let mut attributes = end_attributes(event);
+        if !self.attribute_mappings.is_empty() {
+            let mut projected_attributes = active_span.projected_attributes;
+            projected_attributes.extend(attributes.iter().cloned());
+            attributes.extend(attribute_mapping_aliases(
+                &projected_attributes,
+                &self.attribute_mappings,
+            ));
+        }
+        active_span.span.set_attributes(attributes);
         active_span
             .span
             .end_with_timestamp(to_system_time(*event.timestamp()));
@@ -542,9 +671,13 @@ impl OtelEventProcessor {
         }
         let mark_name = event.name().to_string();
         let timestamp = to_system_time(*event.timestamp());
-        let attributes = mark_attributes(event);
+        let mut attributes = mark_attributes(event);
 
-        if let Some(parent_span) = self.find_parent_span_mut(event) {
+        if self.find_parent_span(event).is_some() {
+            apply_attribute_mappings(&mut attributes, &self.attribute_mappings);
+            let parent_span = self
+                .find_parent_span_mut(event)
+                .expect("parent span was present during mark projection");
             parent_span
                 .span
                 .add_event_with_timestamp(mark_name, timestamp, attributes);
@@ -557,9 +690,9 @@ impl OtelEventProcessor {
             .with_kind(SpanKind::Internal)
             .with_start_time(timestamp)
             .start_with_context(&self.tracer, &self.parent_context(event));
-        let mut span_attributes = attributes;
-        span_attributes.push(KeyValue::new("nemo_relay.mark.orphan", true));
-        span.set_attributes(span_attributes);
+        attributes.push(KeyValue::new("nemo_relay.mark.orphan", true));
+        apply_attribute_mappings(&mut attributes, &self.attribute_mappings);
+        span.set_attributes(attributes);
         span.end_with_timestamp(timestamp);
     }
 
@@ -572,6 +705,7 @@ impl OtelEventProcessor {
         if orphan {
             attributes.push(KeyValue::new("nemo_relay.mark.orphan", true));
         }
+        apply_attribute_mappings(&mut attributes, &self.attribute_mappings);
 
         let mut span = self
             .tracer
@@ -668,37 +802,26 @@ fn scope_type_name(scope_type: Option<ScopeType>) -> &'static str {
 
 fn start_attributes(event: &Event) -> Vec<KeyValue> {
     let mut attributes = common_attributes(event);
-    let handle_attributes = event.attributes();
-    push_serialized(
+    push_serialized_top_level_attributes(
         &mut attributes,
-        "nemo_relay.handle_attributes_json",
-        handle_attributes,
+        "nemo_relay.handle_attributes",
+        event.attributes(),
     );
-    push_serialized(&mut attributes, "nemo_relay.start.data_json", event.data());
-    push_serialized(
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.start.data", event.data());
+    push_top_level_json_attributes(
         &mut attributes,
-        "nemo_relay.start.metadata_json",
+        "nemo_relay.start.metadata",
         event.metadata(),
     );
-    push_serialized(
-        &mut attributes,
-        "nemo_relay.start.input_json",
-        event.input(),
-    );
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.start.input", event.input());
     attributes
 }
 
 fn end_attributes(event: &Event) -> Vec<KeyValue> {
     let mut attributes = Vec::new();
-    push_serialized(&mut attributes, "nemo_relay.end.data_json", event.data());
-
-    let metadata = event.metadata();
-    push_serialized(&mut attributes, "nemo_relay.end.metadata_json", metadata);
-    push_serialized(
-        &mut attributes,
-        "nemo_relay.end.output_json",
-        event.output(),
-    );
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.end.data", event.data());
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.end.metadata", event.metadata());
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.end.output", event.output());
     if event
         .category()
         .is_some_and(|category| category.as_str() == "llm")
@@ -894,7 +1017,6 @@ fn cost_total_and_currency(cost: &CostEstimate) -> Option<(f64, String)> {
 }
 
 fn mark_attributes(event: &Event) -> Vec<KeyValue> {
-    let handle_attributes = event.attributes();
     let mut attributes = vec![
         KeyValue::new("nemo_relay.mark.uuid", event.uuid().to_string()),
         KeyValue::new(
@@ -905,15 +1027,15 @@ fn mark_attributes(event: &Event) -> Vec<KeyValue> {
                 .unwrap_or_default(),
         ),
     ];
-    push_serialized(
+    push_serialized_top_level_attributes(
         &mut attributes,
-        "nemo_relay.mark.attributes_json",
-        handle_attributes,
+        "nemo_relay.mark.attributes",
+        event.attributes(),
     );
-    push_serialized(&mut attributes, "nemo_relay.mark.data_json", event.data());
-    push_serialized(
+    push_top_level_json_attributes(&mut attributes, "nemo_relay.mark.data", event.data());
+    push_top_level_json_attributes(
         &mut attributes,
-        "nemo_relay.mark.metadata_json",
+        "nemo_relay.mark.metadata",
         event.metadata(),
     );
     if let Some(category) = event.category() {
@@ -922,9 +1044,9 @@ fn mark_attributes(event: &Event) -> Vec<KeyValue> {
             category.as_str().to_string(),
         ));
     }
-    push_serialized(
+    push_serialized_top_level_attributes(
         &mut attributes,
-        "nemo_relay.mark.category_profile_json",
+        "nemo_relay.mark.category_profile",
         event.category_profile(),
     );
     attributes
@@ -959,18 +1081,6 @@ fn common_attributes(event: &Event) -> Vec<KeyValue> {
     attributes
 }
 
-fn push_serialized<T: Serialize + ?Sized>(
-    attributes: &mut Vec<KeyValue>,
-    key: &'static str,
-    value: Option<&T>,
-) {
-    if let Some(value) = value
-        && let Ok(json) = serde_json::to_string(value)
-    {
-        attributes.push(KeyValue::new(key, json));
-    }
-}
-
 fn local_parent_span_context(span_context: &SpanContext) -> SpanContext {
     SpanContext::new(
         span_context.trace_id(),
diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs
index 53820bd71..b6999c9f7 100644
--- a/crates/core/src/observability/plugin_component.rs
+++ b/crates/core/src/observability/plugin_component.rs
@@ -54,7 +54,9 @@ use crate::observability::openinference::{
 use crate::observability::otel::{
     OpenTelemetryConfig as CoreOpenTelemetryConfig, OpenTelemetrySubscriber,
 };
-use crate::observability::{MarkProjection, default_mark_exclude_names};
+use crate::observability::{
+    MarkProjection, OtlpAttributeMapping, default_mark_exclude_names, validate_attribute_mappings,
+};
 use crate::plugin::{
     ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, Plugin, PluginComponentSpec, PluginError,
     PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior,
@@ -392,6 +394,9 @@ pub struct OtlpSectionConfig {
     /// Mark names excluded from tool projection. Defaults to `llm.chunk`.
     #[serde(default = "default_mark_exclude_names")]
     pub mark_exclude_names: Vec<String>,
+    /// Typed projected attributes copied to aliases.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub attribute_mappings: Vec<OtlpAttributeMapping>,
     /// OTLP transport: `http_binary` or `grpc`.
     #[serde(default = "default_otlp_transport")]
     #[cfg_attr(feature = "schema", schemars(schema_with = "otlp_transport_schema"))]
@@ -428,6 +433,7 @@ impl Default for OtlpSectionConfig {
             enabled: false,
             mark_projection: MarkProjection::default(),
             mark_exclude_names: default_mark_exclude_names(),
+            attribute_mappings: Vec::new(),
             transport: default_otlp_transport(),
             endpoint: None,
             headers: HashMap::new(),
@@ -564,11 +570,37 @@ static ATOF_SINK_LIST: EditorListItemSpec = EditorListItemSpec {
     list_item: None,
 };
 
+crate::editor_config! {
+    impl OtlpAttributeMapping {
+        key => { label: "key", kind: String },
+        alias => { label: "alias", kind: String },
+    }
+}
+
+fn otlp_attribute_mapping_editor_schema() -> &'static crate::config_editor::EditorSchema {
+    <OtlpAttributeMapping as crate::config_editor::EditorConfig>::editor_schema()
+}
+
+fn default_otlp_attribute_mapping() -> Json {
+    serde_json::to_value(OtlpAttributeMapping::new("", ""))
+        .expect("attribute mapping should serialize")
+}
+
+static OTLP_ATTRIBUTE_MAPPING_LIST_ITEM: crate::config_editor::EditorListItemSpec =
+    crate::config_editor::EditorListItemSpec {
+        kind: crate::config_editor::EditorFieldKind::Section,
+        schema: Some(otlp_attribute_mapping_editor_schema),
+        default: Some(default_otlp_attribute_mapping),
+        tagged_union: None,
+        list_item: None,
+    };
+
 crate::editor_config! {
     impl OtlpSectionConfig {
         enabled => { label: "enabled", kind: Boolean },
         mark_projection => { label: "mark_projection", kind: Enum, values: ["inherit", "event", "tool"] },
         mark_exclude_names => { label: "mark_exclude_names", kind: Json },
+        attribute_mappings => { label: "attribute_mappings", kind: List, list: &OTLP_ATTRIBUTE_MAPPING_LIST_ITEM },
         transport => { label: "transport", kind: Enum, values: ["http_binary", "grpc"] },
         endpoint => { label: "endpoint", kind: String, optional: true },
         headers => { label: "headers", kind: StringMap },
@@ -1463,7 +1495,8 @@ fn build_otel_config(section: OtlpSectionConfig) -> PluginResult<CoreOpenTelemet
     .with_timeout(Duration::from_millis(section.timeout_millis));
     config = config
         .with_mark_projection(section.mark_projection)
-        .with_mark_exclude_names(section.mark_exclude_names);
+        .with_mark_exclude_names(section.mark_exclude_names)
+        .with_attribute_mappings(section.attribute_mappings);
 
     if let Some(endpoint) = section.endpoint {
         config = config.with_endpoint(endpoint);
@@ -1502,7 +1535,8 @@ fn build_openinference_config(section: OtlpSectionConfig) -> PluginResult<CoreOp
         .with_service_name(section.service_name)
         .with_timeout(Duration::from_millis(section.timeout_millis))
         .with_mark_projection(section.mark_projection)
-        .with_mark_exclude_names(section.mark_exclude_names);
+        .with_mark_exclude_names(section.mark_exclude_names)
+        .with_attribute_mappings(section.attribute_mappings);
 
     if let Some(endpoint) = section.endpoint {
         config = config.with_endpoint(endpoint);
@@ -1634,6 +1668,7 @@ fn validate_observability_section_fields(
             "enabled",
             "mark_projection",
             "mark_exclude_names",
+            "attribute_mappings",
             "transport",
             "endpoint",
             "headers",
@@ -1654,6 +1689,7 @@ fn validate_observability_section_fields(
             "enabled",
             "mark_projection",
             "mark_exclude_names",
+            "attribute_mappings",
             "transport",
             "endpoint",
             "headers",
@@ -2388,6 +2424,16 @@ fn validate_otlp_values(
             format!("{section_name} transport must be 'http_binary' or 'grpc'"),
         );
     }
+    if let Err(message) = validate_attribute_mappings(&section.attribute_mappings) {
+        push_policy_diag(
+            diagnostics,
+            policy.unsupported_value,
+            "observability.unsupported_value",
+            Some(section_name.to_string()),
+            Some("attribute_mappings".to_string()),
+            message,
+        );
+    }
 }
 
 fn validate_unknown_fields(
diff --git a/crates/core/tests/integration/middleware_tests.rs b/crates/core/tests/integration/middleware_tests.rs
index e504a6481..b24105e78 100644
--- a/crates/core/tests/integration/middleware_tests.rs
+++ b/crates/core/tests/integration/middleware_tests.rs
@@ -958,10 +958,10 @@ async fn test_managed_tool_pending_marks_project_through_trace_exporters_only()
     for (key, value) in [
         ("nemo_relay.mark.category", "custom"),
         (
-            "nemo_relay.mark.category_profile_json",
-            "{\"subtype\":\"example.compaction\"}",
+            "nemo_relay.mark.category_profile.subtype",
+            "example.compaction",
         ),
-        ("nemo_relay.mark.metadata_json", "{\"source\":\"test\"}"),
+        ("nemo_relay.mark.metadata.source", "test"),
     ] {
         assert!(otel_mark.attributes.iter().any(|attribute| {
             attribute.key.as_str() == key && attribute.value.to_string() == value
@@ -989,10 +989,10 @@ async fn test_managed_tool_pending_marks_project_through_trace_exporters_only()
     for (key, value) in [
         ("nemo_relay.mark.category", "custom"),
         (
-            "nemo_relay.mark.category_profile_json",
-            "{\"subtype\":\"example.compaction\"}",
+            "nemo_relay.mark.category_profile.subtype",
+            "example.compaction",
         ),
-        ("nemo_relay.mark.metadata_json", "{\"source\":\"test\"}"),
+        ("nemo_relay.mark.metadata.source", "test"),
     ] {
         assert!(openinference_mark.attributes.iter().any(|attribute| {
             attribute.key.as_str() == key && attribute.value.to_string() == value
diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs
new file mode 100644
index 000000000..3b6dbc0d3
--- /dev/null
+++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs
@@ -0,0 +1,126 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Unit tests for shared OTLP attribute projection.
+
+use super::{
+    OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_inputs,
+    push_top_level_json_attributes,
+};
+
+#[test]
+fn retains_only_mapping_sources_and_existing_aliases_between_span_events() {
+    let attributes = vec![
+        opentelemetry::KeyValue::new("source", "value"),
+        opentelemetry::KeyValue::new("alias", "existing"),
+        opentelemetry::KeyValue::new("large.request", "payload"),
+    ];
+
+    let retained =
+        attribute_mapping_inputs(&attributes, &[OtlpAttributeMapping::new("source", "alias")]);
+
+    assert_eq!(retained.len(), 2);
+    assert!(
+        retained
+            .iter()
+            .any(|attribute| attribute.key.as_str() == "source")
+    );
+    assert!(
+        retained
+            .iter()
+            .any(|attribute| attribute.key.as_str() == "alias")
+    );
+}
+
+#[test]
+fn projects_typed_json_and_copies_configured_aliases() {
+    let mut attributes = Vec::new();
+    push_top_level_json_attributes(
+        &mut attributes,
+        "nemo_relay.start.metadata",
+        Some(&serde_json::json!({
+            "tenant": "acme",
+            "attempt": 2,
+            "enabled": true,
+            "unset": null,
+            "tags": ["a", "b"],
+            "context": {"region": "us-east-1"},
+            "request": {"id": "nested-id"},
+            "request.id": "flat-id",
+            "event_id": 18446744073709551615u64
+        })),
+    );
+    apply_attribute_mappings(
+        &mut attributes,
+        &[OtlpAttributeMapping::new(
+            "nemo_relay.start.metadata.tenant",
+            "tenant.id",
+        )],
+    );
+
+    assert_eq!(
+        attributes
+            .iter()
+            .find(|attribute| attribute.key.as_str() == "nemo_relay.start.metadata.enabled")
+            .map(|attribute| &attribute.value),
+        Some(&opentelemetry::Value::Bool(true))
+    );
+
+    let values = attributes
+        .iter()
+        .map(|attribute| (attribute.key.as_str(), attribute.value.to_string()))
+        .collect::<std::collections::HashMap<_, _>>();
+    assert_eq!(
+        values.get("nemo_relay.start.metadata.tenant"),
+        Some(&"acme".to_string())
+    );
+    assert_eq!(
+        values.get("nemo_relay.start.metadata.attempt"),
+        Some(&"2".to_string())
+    );
+    assert!(!values.contains_key("nemo_relay.start.metadata.unset"));
+    assert_eq!(
+        values.get("nemo_relay.start.metadata.tags"),
+        Some(&"[\"a\",\"b\"]".to_string())
+    );
+    assert_eq!(
+        values.get("nemo_relay.start.metadata.context"),
+        Some(&"{\"region\":\"us-east-1\"}".to_string())
+    );
+    assert_eq!(
+        values.get("nemo_relay.start.metadata.request"),
+        Some(&"{\"id\":\"nested-id\"}".to_string())
+    );
+    assert_eq!(
+        values.get("nemo_relay.start.metadata.request.id"),
+        Some(&"flat-id".to_string())
+    );
+    assert_eq!(
+        values.get("nemo_relay.start.metadata.event_id"),
+        Some(&"18446744073709551615".to_string())
+    );
+    assert_eq!(values.get("tenant.id"), Some(&"acme".to_string()));
+    assert!(!values.contains_key("nemo_relay.start.metadata_json"));
+}
+
+#[test]
+fn rejects_invalid_attribute_mappings() {
+    assert!(super::validate_attribute_mappings(&[OtlpAttributeMapping::new("", "alias")]).is_err());
+    assert!(
+        super::validate_attribute_mappings(&[
+            OtlpAttributeMapping::new("one", "duplicate"),
+            OtlpAttributeMapping::new("two", "duplicate"),
+        ])
+        .is_err()
+    );
+    assert!(
+        super::validate_attribute_mappings(&[
+            OtlpAttributeMapping::new("one", "duplicate"),
+            OtlpAttributeMapping::new("two", " duplicate "),
+        ])
+        .is_err()
+    );
+    assert!(
+        super::validate_attribute_mappings(&[OtlpAttributeMapping::new("key", "   ")]).is_err()
+    );
+}
diff --git a/crates/core/tests/unit/observability/exporter_parity_tests.rs b/crates/core/tests/unit/observability/exporter_parity_tests.rs
index 16cbbaf3c..4ea93683a 100644
--- a/crates/core/tests/unit/observability/exporter_parity_tests.rs
+++ b/crates/core/tests/unit/observability/exporter_parity_tests.rs
@@ -442,7 +442,10 @@ fn test_usage_facts_parity_for_openai_chat_payload() {
 
     let otel = exports.otel_attrs("model-call");
     assert_no_attribute_key_contains(&otel, "token_count");
-    assert!(otel.contains_key("nemo_relay.end.output_json"));
+    assert!(
+        otel.keys()
+            .any(|key| key.starts_with("nemo_relay.end.output."))
+    );
 }
 
 #[test]
@@ -616,6 +619,8 @@ fn test_tool_call_projection_parity() {
             .get("nemo_relay.tool_call_id"),
         Some(&"call_parity_1".to_string())
     );
+    // OpenTelemetry retains the raw response as JSON; it does not add
+    // OpenInference's semantic tool-call attributes.
     assert_no_attribute_key_contains(&exports.otel_attrs("model-call"), "tool_call");
 }
 
@@ -647,8 +652,20 @@ fn test_reasoning_projected_by_atif_only() {
         Some("I compared both options step by step.")
     );
 
-    assert_no_attribute_key_contains(&exports.otel_attrs("model-call"), "reasoning");
-    assert_no_attribute_key_contains(&exports.openinference_attrs("model-call"), "reasoning");
+    // The raw end data retains reasoning without adding exporter-specific
+    // reasoning semantic conventions.
+    assert_eq!(
+        exports
+            .otel_attrs("model-call")
+            .get("nemo_relay.end.data.reasoning"),
+        Some(&"I compared both options step by step.".to_string())
+    );
+    assert_eq!(
+        exports
+            .openinference_attrs("model-call")
+            .get("nemo_relay.end.data.reasoning"),
+        Some(&"I compared both options step by step.".to_string())
+    );
 }
 
 // ===================================================================
@@ -677,27 +694,35 @@ fn test_replay_payload_preservation_across_exporters() {
         serde_json::from_value(exports.agent_step().extra.clone().unwrap()).unwrap();
     assert_eq!(agent_extra.llm_response, Some(output.clone()));
 
-    // OTel preserves the same payloads as serialized JSON attributes (the
-    // start input keeps the LlmRequest envelope).
+    // OTel projects the LLM request wrapper into typed top-level attributes,
+    // keeping the provider payload in its nested content field.
     let otel = exports.otel_attrs("model-call");
-    let otel_input: Json =
-        serde_json::from_str(otel.get("nemo_relay.start.input_json").unwrap()).unwrap();
+    let otel_input: Json = serde_json::from_str(
+        otel.get("nemo_relay.start.input.content")
+            .expect("projected request content"),
+    )
+    .unwrap();
+    assert_eq!(otel_input, request_content);
     assert_eq!(
-        otel_input,
-        json!({"headers": {}, "content": request_content})
+        otel.get("nemo_relay.start.input.headers"),
+        Some(&"{}".to_string())
+    );
+    assert!(!otel.contains_key("nemo_relay.start.input_json"));
+    assert!(
+        otel.keys()
+            .any(|key| key.starts_with("nemo_relay.end.output."))
     );
-    let otel_output: Json =
-        serde_json::from_str(otel.get("nemo_relay.end.output_json").unwrap()).unwrap();
-    assert_eq!(otel_output, output);
 
-    // OpenInference preserves the raw response payload but omits the raw
-    // LLM request JSON in favor of flattened message attributes plus a
+    // OpenInference preserves the raw response payload but omits raw
+    // LLM request attributes in favor of flattened message fields plus a
     // display input.value.
     let openinference = exports.openinference_attrs("model-call");
-    let openinference_output: Json =
-        serde_json::from_str(openinference.get("nemo_relay.end.output_json").unwrap()).unwrap();
-    assert_eq!(openinference_output, output);
-    assert!(!openinference.contains_key("nemo_relay.start.input_json"));
+    assert!(
+        openinference
+            .keys()
+            .any(|key| key.starts_with("nemo_relay.end.output."))
+    );
+    assert!(!openinference.contains_key("nemo_relay.start.input.content"));
     assert!(openinference.contains_key("llm.input_messages.0.message.content"));
 }
 
diff --git a/crates/core/tests/unit/observability/openinference_tests.rs b/crates/core/tests/unit/observability/openinference_tests.rs
index 0d2f5c107..7f5a03c12 100644
--- a/crates/core/tests/unit/observability/openinference_tests.rs
+++ b/crates/core/tests/unit/observability/openinference_tests.rs
@@ -687,6 +687,133 @@ fn subscriber_registration_and_provider_lifecycle_methods_work() {
     subscriber.shutdown().unwrap();
 }
 
+#[test]
+fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() {
+    let (provider, exporter) = make_provider();
+    let subscriber = OpenInferenceSubscriber::from_tracer_provider_with_options(
+        provider,
+        "mapping-scope",
+        OpenInferenceSubscriberOptions {
+            mark_projection: MarkProjection::Tool,
+            mark_exclude_names: vec!["custom.mark".to_string()],
+            attribute_mappings: vec![
+                crate::observability::OtlpAttributeMapping::new(
+                    "nemo_relay.start.data.tenant",
+                    "tenant.id",
+                ),
+                crate::observability::OtlpAttributeMapping::new(
+                    "nemo_relay.end.data.tenant",
+                    "nemo_relay.start.data.tenant",
+                ),
+                crate::observability::OtlpAttributeMapping::new(
+                    "nemo_relay.start.data.tenant",
+                    "nemo_relay.start.data.existing",
+                ),
+                crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"),
+            ],
+        },
+    )
+    .unwrap();
+    let callback = subscriber.subscriber();
+    let uuid = Uuid::now_v7();
+    callback(&make_start_event(
+        uuid,
+        None,
+        "mapped-scope",
+        ScopeType::Agent,
+        Some(json!({"tenant": 7, "existing": 9})),
+    ));
+    callback(&make_end_event(
+        uuid,
+        None,
+        "mapped-scope",
+        ScopeType::Agent,
+        Some(json!({"tenant": 8})),
+    ));
+    subscriber.force_flush().unwrap();
+
+    let spans = exporter.get_finished_spans().unwrap();
+    let span = spans
+        .iter()
+        .find(|span| span.name.as_ref() == "mapped-scope")
+        .unwrap();
+    assert_eq!(
+        span.attributes
+            .iter()
+            .filter(|attribute| attribute.key.as_str() == "nemo_relay.start.data.tenant")
+            .count(),
+        1
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .find(|attribute| attribute.key.as_str() == "tenant.id")
+            .map(|attribute| &attribute.value),
+        Some(&opentelemetry::Value::I64(7))
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .filter(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing")
+            .count(),
+        1
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .find(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing")
+            .map(|attribute| &attribute.value),
+        Some(&opentelemetry::Value::I64(9))
+    );
+    assert!(
+        !span
+            .attributes
+            .iter()
+            .any(|attribute| attribute.key.as_str() == "ignored.alias")
+    );
+}
+
+#[test]
+fn mapped_orphan_mark_alias_cannot_replace_intrinsic_mark_fields() {
+    let (provider, exporter) = make_provider();
+    let subscriber = OpenInferenceSubscriber::from_tracer_provider_with_attribute_mappings(
+        provider,
+        "mapping-scope",
+        [crate::observability::OtlpAttributeMapping::new(
+            "nemo_relay.mark.data.value",
+            "nemo_relay.mark.orphan",
+        )],
+    )
+    .unwrap();
+    let callback = subscriber.subscriber();
+    callback(&make_mark_event(
+        None,
+        "mapped-orphan-mark",
+        Some(json!({"value": "not-an-orphan-flag"})),
+    ));
+    subscriber.force_flush().unwrap();
+
+    let spans = exporter.get_finished_spans().unwrap();
+    let span = spans
+        .iter()
+        .find(|span| span.name.as_ref() == "mark:mapped-orphan-mark")
+        .unwrap();
+    assert_eq!(
+        span.attributes
+            .iter()
+            .filter(|attribute| attribute.key.as_str() == "nemo_relay.mark.orphan")
+            .count(),
+        1
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .find(|attribute| attribute.key.as_str() == "nemo_relay.mark.orphan")
+            .map(|attribute| &attribute.value),
+        Some(&opentelemetry::Value::Bool(true))
+    );
+}
+
 #[test]
 fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() {
     let _guard = crate::observability::test_mutex().lock().unwrap();
@@ -743,8 +870,8 @@ fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() {
     assert!(!attributes.contains_key("nemo_relay.start.data_json"));
     assert!(!attributes.contains_key("nemo_relay.start.metadata_json"));
     assert_eq!(
-        attributes.get("nemo_relay.start.input_json"),
-        Some(&"{\"task\":\"scope-start\"}".to_string())
+        attributes.get("nemo_relay.start.input.task"),
+        Some(&"scope-start".to_string())
     );
     assert_eq!(
         attributes.get("input.value"),
@@ -755,18 +882,22 @@ fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() {
         Some(&"{\"status\":\"done\"}".to_string())
     );
     assert_eq!(
-        attributes.get("metadata"),
+        attributes.get("openinference.metadata.phase"),
+        Some(&"start".to_string())
+    );
+    assert_eq!(
+        attributes.get(oi::METADATA.as_str()),
         Some(&"{\"phase\":\"start\"}".to_string())
     );
 
     let event_attributes = attr_map(&span.events.events[0].attributes);
     assert_eq!(
-        event_attributes.get("nemo_relay.mark.data_json"),
-        Some(&"{\"step\":1}".to_string())
+        event_attributes.get("nemo_relay.mark.data.step"),
+        Some(&"1".to_string())
     );
     assert_eq!(
-        event_attributes.get("nemo_relay.mark.metadata_json"),
-        Some(&"{\"source\":\"rust-test\"}".to_string())
+        event_attributes.get("nemo_relay.mark.metadata.source"),
+        Some(&"rust-test".to_string())
     );
 }
 
@@ -867,12 +998,12 @@ fn records_span_start_mark_and_end() {
         Some(&root_uuid.to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.start.input_json"),
-        Some(&"{\"query\":\"hello\"}".to_string())
+        attributes.get("nemo_relay.start.input.query"),
+        Some(&"hello".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.end.output_json"),
-        Some(&"{\"result\":\"ok\"}".to_string())
+        attributes.get("nemo_relay.end.output.result"),
+        Some(&"ok".to_string())
     );
 }
 
@@ -942,21 +1073,13 @@ fn openclaw_model_timing_marks_attach_to_parent_spans() {
         ambiguous_attributes.get("nemo_relay.mark.parent_uuid"),
         Some(&root_uuid.to_string())
     );
-    let ambiguous_data: serde_json::Value = serde_json::from_str(
-        ambiguous_attributes
-            .get("nemo_relay.mark.data_json")
-            .unwrap(),
-    )
-    .unwrap();
     assert_eq!(
-        ambiguous_data,
-        json!({
-            "runId": "run-1",
-            "sessionId": "session-1",
-            "provider": "openai",
-            "model": "gpt-4",
-            "candidateCount": 2
-        })
+        ambiguous_attributes.get("nemo_relay.mark.data.runId"),
+        Some(&"run-1".to_string())
+    );
+    assert_eq!(
+        ambiguous_attributes.get("nemo_relay.mark.data.candidateCount"),
+        Some(&"2".to_string())
     );
     assert!(!ambiguous_attributes.contains_key("nemo_relay.mark.metadata_json"));
 
@@ -965,22 +1088,13 @@ fn openclaw_model_timing_marks_attach_to_parent_spans() {
         unpaired_attributes.get("nemo_relay.mark.parent_uuid"),
         Some(&root_uuid.to_string())
     );
-    let unpaired_data: serde_json::Value = serde_json::from_str(
-        unpaired_attributes
-            .get("nemo_relay.mark.data_json")
-            .unwrap(),
-    )
-    .unwrap();
     assert_eq!(
-        unpaired_data,
-        json!({
-            "runId": "run-1",
-            "callId": "call-1",
-            "provider": "openai",
-            "model": "gpt-4",
-            "durationMs": 42,
-            "outcome": "completed"
-        })
+        unpaired_attributes.get("nemo_relay.mark.data.runId"),
+        Some(&"run-1".to_string())
+    );
+    assert_eq!(
+        unpaired_attributes.get("nemo_relay.mark.data.durationMs"),
+        Some(&"42".to_string())
     );
     assert!(!unpaired_attributes.contains_key("nemo_relay.mark.metadata_json"));
 }
@@ -1159,7 +1273,7 @@ fn llm_input_value_omits_request_headers() {
     let attributes = attr_map(&spans[0].attributes);
     assert_attr(&attributes, "input.value", "user: hi");
     assert_attr(&attributes, "input.mime_type", "text/plain");
-    assert!(!attributes.contains_key("nemo_relay.start.input_json"));
+    assert!(!attributes.contains_key("nemo_relay.start.input.content"));
     assert!(!attributes["input.value"].contains("authorization"));
     assert!(!attributes["input.value"].contains("secret-token"));
     // The provider-shaped request is decoded through the codec layer, so
@@ -1867,11 +1981,12 @@ fn output_value_prefers_display_content() {
         Some(&"text/plain".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.end.output_json"),
-        Some(
-            &"{\"content\":\"Tool edit completed.\",\"details\":{\"diff\":\"-old\\n+new\"}}"
-                .to_string()
-        )
+        attributes.get("nemo_relay.end.output.content"),
+        Some(&"Tool edit completed.".to_string())
+    );
+    assert_eq!(
+        attributes.get("nemo_relay.end.output.details"),
+        Some(&"{\"diff\":\"-old\\n+new\"}".to_string())
     );
     assert!(!attributes.contains_key("nemo_relay.end.data_json"));
 }
@@ -2304,16 +2419,16 @@ fn tool_projection_emits_generic_mark_as_parented_openinference_tool_span() {
         Some(&"tool".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.mark.data_json"),
-        Some(&"{\"count\":3}".to_string())
+        attributes.get("nemo_relay.mark.data.count"),
+        Some(&"3".to_string())
     );
     assert_eq!(
         attributes.get("nemo_relay.mark.category"),
         Some(&"custom".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.mark.category_profile_json"),
-        Some(&"{\"subtype\":\"example.compaction\"}".to_string())
+        attributes.get("nemo_relay.mark.category_profile.subtype"),
+        Some(&"example.compaction".to_string())
     );
     assert!(!attributes.contains_key("nemo_relay.mark.orphan"));
 }
@@ -2675,6 +2790,12 @@ fn semantic_scope_type_and_input_value_follow_event_variants() {
     );
     assert_eq!(semantic_scope_type(&remote_tool), Some(ScopeType::Tool));
     assert_eq!(span_kind(&remote_tool), SpanKind::Client);
+    let remote_tool_attributes = attr_map(&start_attributes(&remote_tool));
+    assert_eq!(
+        remote_tool_attributes.get("nemo_relay.handle_attributes"),
+        Some(&"[\"remote\"]".to_string())
+    );
+    assert!(!remote_tool_attributes.contains_key("nemo_relay.handle_attributes_json"));
     let (remote_tool_input, remote_tool_mime_type) =
         openinference_input_value(&remote_tool).unwrap();
     assert_eq!(remote_tool_mime_type, "application/json");
@@ -2716,11 +2837,12 @@ fn scope_end_output_payload_is_exported_to_openinference_attributes() {
         json!({"status": "done", "metrics": {"tokens": 42}})
     );
     assert_eq!(
-        serde_json::from_str::<serde_json::Value>(
-            attributes.get("nemo_relay.end.output_json").unwrap(),
-        )
-        .unwrap(),
-        json!({"status": "done", "metrics": {"tokens": 42}})
+        attributes.get("nemo_relay.end.output.status"),
+        Some(&"done".to_string())
+    );
+    assert_eq!(
+        attributes.get("nemo_relay.end.output.metrics"),
+        Some(&"{\"tokens\":42}".to_string())
     );
 }
 
@@ -2857,6 +2979,10 @@ fn helper_functions_cover_additional_openinference_branches() {
         raw_model_attributes.get(oi::llm::MODEL_NAME.as_str()),
         Some(&"raw-model".to_string())
     );
+    assert_eq!(
+        llm_attributes.get("openinference.metadata.phase"),
+        Some(&"done".to_string())
+    );
     assert_eq!(
         llm_attributes.get(oi::METADATA.as_str()),
         Some(&"{\"phase\":\"done\"}".to_string())
@@ -2928,12 +3054,12 @@ fn helper_functions_cover_additional_openinference_branches() {
     ));
     let mark_attributes = attr_map(&mark_attributes(&mark));
     assert_eq!(
-        mark_attributes.get("nemo_relay.mark.data_json"),
-        Some(&"{\"kind\":\"aux\"}".to_string())
+        mark_attributes.get("nemo_relay.mark.data.kind"),
+        Some(&"aux".to_string())
     );
     assert_eq!(
-        mark_attributes.get("nemo_relay.mark.metadata_json"),
-        Some(&"{\"source\":\"unit\"}".to_string())
+        mark_attributes.get("nemo_relay.mark.metadata.source"),
+        Some(&"unit".to_string())
     );
 
     let llm_with_scalar_input = make_start_event(
@@ -3840,11 +3966,13 @@ fn hermes_exact_api_payloads_emit_openinference_text_usage_and_metadata() {
         attributes.get("llm.cost.total"),
         Some(&"0.0042".to_string())
     );
-    assert_attr_contains(&attributes, "metadata", "\"provider_payload_exact\":true");
-    assert_attr_contains(
-        &attributes,
-        "metadata",
-        "\"fidelity_source\":\"hermes_api_hooks_sanitized\"",
+    assert_eq!(
+        attributes.get("openinference.metadata.provider_payload_exact"),
+        Some(&"true".to_string())
+    );
+    assert_eq!(
+        attributes.get("openinference.metadata.fidelity_source"),
+        Some(&"hermes_api_hooks_sanitized".to_string())
     );
 }
 
@@ -3934,26 +4062,25 @@ fn hermes_api_request_error_emits_openinference_json_output_and_metadata() {
         Some(&"application/json".to_string())
     );
     assert_eq!(
-        serde_json::from_str::<serde_json::Value>(
-            attributes.get("nemo_relay.end.output_json").unwrap(),
-        )
-        .unwrap(),
-        json!({
-            "status_code": 502,
-            "retry_count": 1,
-            "max_retries": 2,
-            "retryable": true,
-            "reason": "upstream",
-            "error": {
-                "type": "BadGateway",
-                "message": "gateway upstream error"
-            }
-        })
+        attributes.get("nemo_relay.end.output.status_code"),
+        Some(&"502".to_string())
+    );
+    assert_eq!(
+        attributes.get("openinference.metadata.provider_payload_exact"),
+        Some(&"false".to_string())
+    );
+    assert_eq!(
+        attributes.get("openinference.metadata.fidelity_source"),
+        Some(&"hermes_api_hooks".to_string())
+    );
+    assert_attr_contains(
+        &attributes,
+        oi::METADATA.as_str(),
+        "\"provider_payload_exact\":false",
     );
-    assert_attr_contains(&attributes, "metadata", "\"provider_payload_exact\":false");
     assert_attr_contains(
         &attributes,
-        "metadata",
+        oi::METADATA.as_str(),
         "\"fidelity_source\":\"hermes_api_hooks\"",
     );
 }
diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs
index b764c29a6..a5b790ccb 100644
--- a/crates/core/tests/unit/observability/otel_tests.rs
+++ b/crates/core/tests/unit/observability/otel_tests.rs
@@ -584,6 +584,133 @@ fn subscriber_registration_and_provider_lifecycle_methods_work() {
     subscriber.shutdown().unwrap();
 }
 
+#[test]
+fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() {
+    let (provider, exporter) = make_provider();
+    let subscriber = OpenTelemetrySubscriber::from_tracer_provider_with_options(
+        provider,
+        "mapping-scope",
+        OpenTelemetrySubscriberOptions {
+            mark_projection: MarkProjection::Tool,
+            mark_exclude_names: vec!["custom.mark".to_string()],
+            attribute_mappings: vec![
+                crate::observability::OtlpAttributeMapping::new(
+                    "nemo_relay.start.data.tenant",
+                    "tenant.id",
+                ),
+                crate::observability::OtlpAttributeMapping::new(
+                    "nemo_relay.end.data.tenant",
+                    "nemo_relay.start.data.tenant",
+                ),
+                crate::observability::OtlpAttributeMapping::new(
+                    "nemo_relay.start.data.tenant",
+                    "nemo_relay.start.data.existing",
+                ),
+                crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"),
+            ],
+        },
+    )
+    .unwrap();
+    let callback = subscriber.subscriber();
+    let uuid = Uuid::now_v7();
+    callback(&make_start_event(
+        uuid,
+        None,
+        "mapped-scope",
+        ScopeType::Agent,
+        Some(json!({"tenant": 7, "existing": 9})),
+    ));
+    callback(&make_end_event(
+        uuid,
+        None,
+        "mapped-scope",
+        ScopeType::Agent,
+        Some(json!({"tenant": 8})),
+    ));
+    subscriber.force_flush().unwrap();
+
+    let spans = exporter.get_finished_spans().unwrap();
+    let span = spans
+        .iter()
+        .find(|span| span.name.as_ref() == "mapped-scope")
+        .unwrap();
+    assert_eq!(
+        span.attributes
+            .iter()
+            .filter(|attribute| attribute.key.as_str() == "nemo_relay.start.data.tenant")
+            .count(),
+        1
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .find(|attribute| attribute.key.as_str() == "tenant.id")
+            .map(|attribute| &attribute.value),
+        Some(&opentelemetry::Value::I64(7))
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .filter(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing")
+            .count(),
+        1
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .find(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing")
+            .map(|attribute| &attribute.value),
+        Some(&opentelemetry::Value::I64(9))
+    );
+    assert!(
+        !span
+            .attributes
+            .iter()
+            .any(|attribute| attribute.key.as_str() == "ignored.alias")
+    );
+}
+
+#[test]
+fn mapped_orphan_mark_alias_cannot_replace_intrinsic_mark_fields() {
+    let (provider, exporter) = make_provider();
+    let subscriber = OpenTelemetrySubscriber::from_tracer_provider_with_attribute_mappings(
+        provider,
+        "mapping-scope",
+        [crate::observability::OtlpAttributeMapping::new(
+            "nemo_relay.mark.data.value",
+            "nemo_relay.mark.orphan",
+        )],
+    )
+    .unwrap();
+    let callback = subscriber.subscriber();
+    callback(&make_mark_event(
+        None,
+        "mapped-orphan-mark",
+        Some(json!({"value": "not-an-orphan-flag"})),
+    ));
+    subscriber.force_flush().unwrap();
+
+    let spans = exporter.get_finished_spans().unwrap();
+    let span = spans
+        .iter()
+        .find(|span| span.name.as_ref() == "mark:mapped-orphan-mark")
+        .unwrap();
+    assert_eq!(
+        span.attributes
+            .iter()
+            .filter(|attribute| attribute.key.as_str() == "nemo_relay.mark.orphan")
+            .count(),
+        1
+    );
+    assert_eq!(
+        span.attributes
+            .iter()
+            .find(|attribute| attribute.key.as_str() == "nemo_relay.mark.orphan")
+            .map(|attribute| &attribute.value),
+        Some(&opentelemetry::Value::Bool(true))
+    );
+}
+
 #[test]
 fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() {
     let _guard = crate::observability::test_mutex().lock().unwrap();
@@ -634,22 +761,22 @@ fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() {
 
     let attributes = attr_map(&span.attributes);
     assert_eq!(
-        attributes.get("nemo_relay.start.data_json"),
-        Some(&"{\"task\":\"scope-start\"}".to_string())
+        attributes.get("nemo_relay.start.input.task"),
+        Some(&"scope-start".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.start.metadata_json"),
-        Some(&"{\"phase\":\"start\"}".to_string())
+        attributes.get("nemo_relay.start.metadata.phase"),
+        Some(&"start".to_string())
     );
 
     let event_attributes = attr_map(&span.events.events[0].attributes);
     assert_eq!(
-        event_attributes.get("nemo_relay.mark.data_json"),
-        Some(&"{\"step\":1}".to_string())
+        event_attributes.get("nemo_relay.mark.data.step"),
+        Some(&"1".to_string())
     );
     assert_eq!(
-        event_attributes.get("nemo_relay.mark.metadata_json"),
-        Some(&"{\"source\":\"rust-test\"}".to_string())
+        event_attributes.get("nemo_relay.mark.metadata.source"),
+        Some(&"rust-test".to_string())
     );
 }
 
@@ -747,12 +874,12 @@ fn records_span_start_mark_and_end() {
         Some(&root_uuid.to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.start.input_json"),
-        Some(&"{\"query\":\"hello\"}".to_string())
+        attributes.get("nemo_relay.start.input.query"),
+        Some(&"hello".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.end.output_json"),
-        Some(&"{\"result\":\"ok\"}".to_string())
+        attributes.get("nemo_relay.end.output.result"),
+        Some(&"ok".to_string())
     );
 }
 
@@ -977,16 +1104,16 @@ fn tool_projection_emits_generic_mark_as_parented_zero_duration_span() {
         Some(&"tool".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.mark.data_json"),
-        Some(&"{\"count\":3}".to_string())
+        attributes.get("nemo_relay.mark.data.count"),
+        Some(&"3".to_string())
     );
     assert_eq!(
         attributes.get("nemo_relay.mark.category"),
         Some(&"custom".to_string())
     );
     assert_eq!(
-        attributes.get("nemo_relay.mark.category_profile_json"),
-        Some(&"{\"subtype\":\"example.compaction\"}".to_string())
+        attributes.get("nemo_relay.mark.category_profile.subtype"),
+        Some(&"example.compaction".to_string())
     );
     assert!(!attributes.contains_key("nemo_relay.mark.orphan"));
 }
@@ -1240,6 +1367,12 @@ fn semantic_scope_type_and_span_kind_follow_event_variants() {
     );
     assert_eq!(semantic_scope_type(&remote_tool), Some(ScopeType::Tool));
     assert_eq!(span_kind(&remote_tool), SpanKind::Client);
+    let remote_tool_attributes = attr_map(&start_attributes(&remote_tool));
+    assert_eq!(
+        remote_tool_attributes.get("nemo_relay.handle_attributes"),
+        Some(&"[\"remote\"]".to_string())
+    );
+    assert!(!remote_tool_attributes.contains_key("nemo_relay.handle_attributes_json"));
 
     let llm_event = make_end_event(
         Uuid::now_v7(),
@@ -1328,7 +1461,7 @@ fn llm_end_emits_cost_only_no_token_or_gen_ai_attributes() {
     assert!(keys.iter().any(|k| k == "nemo_relay.llm.cost.currency"));
     assert!(
         keys.iter()
-            .all(|k| !k.to_ascii_lowercase().contains("token") && !k.starts_with("gen_ai")),
+            .all(|k| !k.starts_with("llm.token") && !k.starts_with("gen_ai")),
         "no token attributes expected on the LLM span: {keys:?}"
     );
 }
@@ -1453,12 +1586,12 @@ fn helper_functions_cover_additional_otel_branches() {
 
     let start_attributes = attr_map(&start_attributes(&tool_event));
     assert_eq!(
-        start_attributes.get("nemo_relay.start.input_json"),
-        Some(&"{\"query\":\"hello\"}".to_string())
+        start_attributes.get("nemo_relay.start.data.query"),
+        Some(&"hello".to_string())
     );
     assert_eq!(
-        start_attributes.get("nemo_relay.start.metadata_json"),
-        Some(&"{\"meta\":true}".to_string())
+        start_attributes.get("nemo_relay.start.metadata.meta"),
+        Some(&"true".to_string())
     );
 
     let tool_end_attributes = attr_map(&end_attributes(&Event::Scope(ScopeEvent::new(
@@ -1473,8 +1606,8 @@ fn helper_functions_cover_additional_otel_branches() {
         Some(CategoryProfile::builder().tool_call_id("call-456").build()),
     ))));
     assert_eq!(
-        tool_end_attributes.get("nemo_relay.end.output_json"),
-        Some(&"{\"result\":true}".to_string())
+        tool_end_attributes.get("nemo_relay.end.data.result"),
+        Some(&"true".to_string())
     );
 
     {
@@ -1767,12 +1900,12 @@ fn helper_functions_cover_additional_otel_branches() {
     ));
     let mark_attributes = attr_map(&mark_attributes(&mark));
     assert_eq!(
-        mark_attributes.get("nemo_relay.mark.data_json"),
-        Some(&"{\"kind\":\"aux\"}".to_string())
+        mark_attributes.get("nemo_relay.mark.data.kind"),
+        Some(&"aux".to_string())
     );
     assert_eq!(
-        mark_attributes.get("nemo_relay.mark.metadata_json"),
-        Some(&"{\"source\":\"unit\"}".to_string())
+        mark_attributes.get("nemo_relay.mark.metadata.source"),
+        Some(&"unit".to_string())
     );
 
     let mut processor = OtelEventProcessor::new(make_provider().0, "test".into());
diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs
index cd4bebef2..d29d29238 100644
--- a/crates/core/tests/unit/observability/plugin_component_tests.rs
+++ b/crates/core/tests/unit/observability/plugin_component_tests.rs
@@ -144,6 +144,18 @@ fn editor_schema_tracks_observability_config_types() {
         .expect("openinference editor schema");
     let headers = otlp.field("headers").expect("headers field");
     assert_eq!(headers.kind, EditorFieldKind::StringMap);
+
+    let attribute_mappings = otlp
+        .field("attribute_mappings")
+        .expect("attribute mappings field");
+    assert_eq!(attribute_mappings.kind, EditorFieldKind::List);
+    let mapping = attribute_mappings
+        .list_item
+        .expect("attribute mapping list item");
+    assert_eq!(mapping.kind, EditorFieldKind::Section);
+    let mapping_schema = mapping.schema.expect("attribute mapping schema")();
+    assert_eq!(mapping_schema.fields[0].name, "key");
+    assert_eq!(mapping_schema.fields[1].name, "alias");
 }
 
 fn push_agent(name: &str) -> crate::api::scope::ScopeHandle {
@@ -286,6 +298,7 @@ fn default_config_and_component_conversion_cover_public_shape() {
     assert!(!otlp.enabled);
     assert_eq!(otlp.mark_projection, MarkProjection::Inherit);
     assert_eq!(otlp.mark_exclude_names, vec!["llm.chunk"]);
+    assert!(otlp.attribute_mappings.is_empty());
     assert_eq!(otlp.transport, "http_binary");
     assert_eq!(otlp.service_name, "nemo-relay");
     assert_eq!(otlp.timeout_millis, 3_000);
@@ -306,6 +319,15 @@ fn default_config_and_component_conversion_cover_public_shape() {
 
 #[test]
 fn mark_projection_parses_for_otlp_and_rejects_unknown_values() {
+    let mappings: OtlpSectionConfig = serde_json::from_value(json!({
+        "attribute_mappings": [{
+            "key": "nemo_relay.start.metadata.tenant",
+            "alias": "tenant.id"
+        }]
+    }))
+    .unwrap();
+    assert_eq!(mappings.attribute_mappings.len(), 1);
+
     let otlp: OtlpSectionConfig = serde_json::from_value(json!({
         "mark_projection": "tool"
     }))
@@ -351,6 +373,43 @@ fn mark_projection_parses_for_otlp_and_rejects_unknown_values() {
         diagnostic.code == "observability.unknown_field"
             && diagnostic.field.as_deref() == Some("mark_projection")
     }));
+
+    let report = validate_plugin_config(&plugin_config(json!({
+        "openinference": {
+            "attribute_mappings": [
+                {"key": "", "alias": "tenant.id"},
+                {"key": "openinference.metadata.tenant", "alias": "tenant.id"}
+            ]
+        }
+    })));
+    assert!(report.diagnostics.iter().any(|diagnostic| {
+        diagnostic.code == "observability.unsupported_value"
+            && diagnostic.field.as_deref() == Some("attribute_mappings")
+            && diagnostic
+                .message
+                .contains("attribute mapping key must not be blank")
+    }));
+
+    let report = validate_plugin_config(&plugin_config(json!({
+        "policy": {"unknown_field": "error"},
+        "opentelemetry": {
+            "attribute_mappings": [{
+                "key": "nemo_relay.start.metadata.tenant",
+                "alias": "tenant.id"
+            }]
+        },
+        "openinference": {
+            "attribute_mappings": [{
+                "key": "openinference.metadata.tenant",
+                "alias": "tenant.id"
+            }]
+        }
+    })));
+    assert!(
+        !report.has_errors(),
+        "valid attribute mappings must not be reported as unknown: {:?}",
+        report.diagnostics
+    );
 }
 
 #[cfg(feature = "schema")]
@@ -378,6 +437,7 @@ fn schema_contains_every_supported_observability_option() {
         "model_name",
         "mark_projection",
         "mark_exclude_names",
+        "attribute_mappings",
         "tool_definitions",
         "extra",
         "filename_template",
diff --git a/crates/ffi/nemo_relay.h b/crates/ffi/nemo_relay.h
index 2b053d49b..652d6c108 100644
--- a/crates/ffi/nemo_relay.h
+++ b/crates/ffi/nemo_relay.h
@@ -1329,6 +1329,26 @@ NemoRelayStatus nemo_relay_otel_subscriber_create(const char *transport,
                                                   uint64_t timeout_millis,
                                                   struct FfiOpenTelemetrySubscriber **out);
 
+/**
+ * Creates a new OpenTelemetry subscriber with typed attribute mappings.
+ *
+ * `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects.
+ *
+ * # Safety
+ * Any non-null C strings must be valid and `out` must be non-null.
+ */
+NemoRelayStatus nemo_relay_otel_subscriber_create_with_attribute_mappings(const char *transport,
+                                                                          const char *endpoint,
+                                                                          const char *headers_json,
+                                                                          const char *resource_attributes_json,
+                                                                          const char *service_name,
+                                                                          const char *service_namespace,
+                                                                          const char *service_version,
+                                                                          const char *instrumentation_scope,
+                                                                          uint64_t timeout_millis,
+                                                                          const char *attribute_mappings_json,
+                                                                          struct FfiOpenTelemetrySubscriber **out);
+
 /**
  * Registers the OpenTelemetry subscriber as an event subscriber.
  *
@@ -1383,6 +1403,26 @@ NemoRelayStatus nemo_relay_openinference_subscriber_create(const char *transport
                                                            uint64_t timeout_millis,
                                                            struct FfiOpenInferenceSubscriber **out);
 
+/**
+ * Creates a new OpenInference subscriber with typed attribute mappings.
+ *
+ * `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects.
+ *
+ * # Safety
+ * Any non-null C strings must be valid and `out` must be non-null.
+ */
+NemoRelayStatus nemo_relay_openinference_subscriber_create_with_attribute_mappings(const char *transport,
+                                                                                   const char *endpoint,
+                                                                                   const char *headers_json,
+                                                                                   const char *resource_attributes_json,
+                                                                                   const char *service_name,
+                                                                                   const char *service_namespace,
+                                                                                   const char *service_version,
+                                                                                   const char *instrumentation_scope,
+                                                                                   uint64_t timeout_millis,
+                                                                                   const char *attribute_mappings_json,
+                                                                                   struct FfiOpenInferenceSubscriber **out);
+
 /**
  * Registers the OpenInference subscriber as an event subscriber.
  *
diff --git a/crates/ffi/src/api/observability.rs b/crates/ffi/src/api/observability.rs
index 61ba1eb50..989b4f127 100644
--- a/crates/ffi/src/api/observability.rs
+++ b/crates/ffi/src/api/observability.rs
@@ -16,6 +16,7 @@ type OpenTelemetryConfig = nemo_relay::observability::otel::OpenTelemetryConfig;
 type OpenTelemetrySubscriber = nemo_relay::observability::otel::OpenTelemetrySubscriber;
 type OpenInferenceConfig = nemo_relay::observability::openinference::OpenInferenceConfig;
 type OpenInferenceSubscriber = nemo_relay::observability::openinference::OpenInferenceSubscriber;
+type OtlpAttributeMapping = nemo_relay::observability::OtlpAttributeMapping;
 type ObservabilityComponentSpec = nemo_relay::observability::plugin_component::ComponentSpec;
 type ObservabilityConfig = nemo_relay::observability::plugin_component::ObservabilityConfig;
 
@@ -533,6 +534,28 @@ fn parse_string_map_json(
     Ok(out)
 }
 
+fn parse_attribute_mappings_json(
+    json_ptr: *const c_char,
+) -> Result<Vec<OtlpAttributeMapping>, NemoRelayStatus> {
+    if json_ptr.is_null() {
+        return Ok(Vec::new());
+    }
+    let json_string = c_str_to_string(json_ptr)?;
+    let value: serde_json::Value = serde_json::from_str(&json_string).map_err(|error| {
+        set_last_error(&format!("invalid attribute_mappings JSON: {error}"));
+        NemoRelayStatus::InvalidJson
+    })?;
+    let mappings: Vec<OtlpAttributeMapping> = serde_json::from_value(value).map_err(|error| {
+        set_last_error(&format!("invalid attribute_mappings: {error}"));
+        NemoRelayStatus::InvalidArg
+    })?;
+    nemo_relay::observability::validate_attribute_mappings(&mappings).map_err(|error| {
+        set_last_error(&format!("invalid attribute_mappings: {error}"));
+        NemoRelayStatus::InvalidArg
+    })?;
+    Ok(mappings)
+}
+
 fn required_out_ptr<T>(out: *mut *mut T) -> Result<(), NemoRelayStatus> {
     if out.is_null() {
         set_last_error("out pointer is null");
@@ -670,6 +693,43 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create(
     instrumentation_scope: *const c_char,
     timeout_millis: u64,
     out: *mut *mut FfiOpenTelemetrySubscriber,
+) -> NemoRelayStatus {
+    unsafe {
+        nemo_relay_otel_subscriber_create_with_attribute_mappings(
+            transport,
+            endpoint,
+            headers_json,
+            resource_attributes_json,
+            service_name,
+            service_namespace,
+            service_version,
+            instrumentation_scope,
+            timeout_millis,
+            std::ptr::null(),
+            out,
+        )
+    }
+}
+
+/// Creates a new OpenTelemetry subscriber with typed attribute mappings.
+///
+/// `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects.
+///
+/// # Safety
+/// Any non-null C strings must be valid and `out` must be non-null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_attribute_mappings(
+    transport: *const c_char,
+    endpoint: *const c_char,
+    headers_json: *const c_char,
+    resource_attributes_json: *const c_char,
+    service_name: *const c_char,
+    service_namespace: *const c_char,
+    service_version: *const c_char,
+    instrumentation_scope: *const c_char,
+    timeout_millis: u64,
+    attribute_mappings_json: *const c_char,
+    out: *mut *mut FfiOpenTelemetrySubscriber,
 ) -> NemoRelayStatus {
     clear_last_error();
     if let Err(status) = required_out_ptr(out) {
@@ -727,6 +787,11 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create(
         Ok(config) => config,
         Err(status) => return status,
     };
+    let attribute_mappings = match parse_attribute_mappings_json(attribute_mappings_json) {
+        Ok(mappings) => mappings,
+        Err(status) => return status,
+    };
+    config = config.with_attribute_mappings(attribute_mappings);
     config = match apply_string_map(
         config,
         resource_attributes_json,
@@ -859,6 +924,43 @@ pub unsafe extern "C" fn nemo_relay_openinference_subscriber_create(
     instrumentation_scope: *const c_char,
     timeout_millis: u64,
     out: *mut *mut FfiOpenInferenceSubscriber,
+) -> NemoRelayStatus {
+    unsafe {
+        nemo_relay_openinference_subscriber_create_with_attribute_mappings(
+            transport,
+            endpoint,
+            headers_json,
+            resource_attributes_json,
+            service_name,
+            service_namespace,
+            service_version,
+            instrumentation_scope,
+            timeout_millis,
+            std::ptr::null(),
+            out,
+        )
+    }
+}
+
+/// Creates a new OpenInference subscriber with typed attribute mappings.
+///
+/// `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects.
+///
+/// # Safety
+/// Any non-null C strings must be valid and `out` must be non-null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn nemo_relay_openinference_subscriber_create_with_attribute_mappings(
+    transport: *const c_char,
+    endpoint: *const c_char,
+    headers_json: *const c_char,
+    resource_attributes_json: *const c_char,
+    service_name: *const c_char,
+    service_namespace: *const c_char,
+    service_version: *const c_char,
+    instrumentation_scope: *const c_char,
+    timeout_millis: u64,
+    attribute_mappings_json: *const c_char,
+    out: *mut *mut FfiOpenInferenceSubscriber,
 ) -> NemoRelayStatus {
     clear_last_error();
     if let Err(status) = required_out_ptr(out) {
@@ -916,6 +1018,11 @@ pub unsafe extern "C" fn nemo_relay_openinference_subscriber_create(
         Ok(config) => config,
         Err(status) => return status,
     };
+    let attribute_mappings = match parse_attribute_mappings_json(attribute_mappings_json) {
+        Ok(mappings) => mappings,
+        Err(status) => return status,
+    };
+    config = config.with_attribute_mappings(attribute_mappings);
     config = match apply_string_map(
         config,
         resource_attributes_json,
diff --git a/crates/ffi/tests/unit/api/plugin_tests.rs b/crates/ffi/tests/unit/api/plugin_tests.rs
index 26c75e904..ebf6a4217 100644
--- a/crates/ffi/tests/unit/api/plugin_tests.rs
+++ b/crates/ffi/tests/unit/api/plugin_tests.rs
@@ -1466,6 +1466,123 @@ fn test_ffi_specialized_subscriber_and_exporter_default_and_invalid_name_paths()
     }
 }
 
+#[test]
+fn test_ffi_typed_attribute_mapping_constructors_validate_and_accept_mappings() {
+    let _lock = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
+    reset_globals();
+
+    unsafe {
+        let valid = cstring(r#"[{"key":"nemo_relay.start.data.tenant","alias":"tenant.id"}]"#);
+        let invalid = cstring(r#"[{"key":"","alias":"tenant.id"}]"#);
+
+        let mut otel = ptr::null_mut();
+        assert_eq!(
+            nemo_relay_otel_subscriber_create_with_attribute_mappings(
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                0,
+                valid.as_ptr(),
+                &mut otel,
+            ),
+            NemoRelayStatus::Ok
+        );
+        nemo_relay_otel_subscriber_free(otel);
+        assert_eq!(
+            nemo_relay_otel_subscriber_create_with_attribute_mappings(
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                0,
+                invalid.as_ptr(),
+                &mut otel,
+            ),
+            NemoRelayStatus::InvalidArg
+        );
+
+        let mut openinference = ptr::null_mut();
+        assert_eq!(
+            nemo_relay_openinference_subscriber_create_with_attribute_mappings(
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                0,
+                valid.as_ptr(),
+                &mut openinference,
+            ),
+            NemoRelayStatus::Ok
+        );
+        nemo_relay_openinference_subscriber_free(openinference);
+        assert_eq!(
+            nemo_relay_openinference_subscriber_create_with_attribute_mappings(
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                ptr::null(),
+                0,
+                invalid.as_ptr(),
+                &mut openinference,
+            ),
+            NemoRelayStatus::InvalidArg
+        );
+
+        for invalid_shape in ["{}", "null", r#"[{"key":1,"alias":"tenant.id"}]"#] {
+            let invalid_shape = cstring(invalid_shape);
+            assert_eq!(
+                nemo_relay_otel_subscriber_create_with_attribute_mappings(
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    0,
+                    invalid_shape.as_ptr(),
+                    &mut otel,
+                ),
+                NemoRelayStatus::InvalidArg
+            );
+            assert_eq!(
+                nemo_relay_openinference_subscriber_create_with_attribute_mappings(
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    ptr::null(),
+                    0,
+                    invalid_shape.as_ptr(),
+                    &mut openinference,
+                ),
+                NemoRelayStatus::InvalidArg
+            );
+        }
+    }
+}
+
 #[test]
 fn test_ffi_specialized_constructor_invalid_utf8_and_malformed_json_sweep() {
     let _lock = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
diff --git a/crates/node/observability.d.ts b/crates/node/observability.d.ts
index d083ae7ed..d85cd1d1f 100644
--- a/crates/node/observability.d.ts
+++ b/crates/node/observability.d.ts
@@ -69,6 +69,7 @@ export interface OtlpConfig {
   enabled?: boolean;
   mark_projection?: 'inherit' | 'event' | 'tool';
   mark_exclude_names?: string[];
+  attribute_mappings?: OtlpAttributeMapping[];
   transport?: 'http_binary' | 'grpc' | string;
   endpoint?: string;
   headers?: Record<string, string>;
@@ -80,6 +81,12 @@ export interface OtlpConfig {
   timeout_millis?: number;
 }
 
+/** Copy a projected OTLP attribute to an additional attribute name. */
+export interface OtlpAttributeMapping {
+  key: string;
+  alias: string;
+}
+
 export interface Config {
   version?: number;
   atof?: AtofConfig;
diff --git a/crates/node/observability.js b/crates/node/observability.js
index 72a814c10..a43d69e9c 100644
--- a/crates/node/observability.js
+++ b/crates/node/observability.js
@@ -58,6 +58,7 @@ function otlpConfig(config = {}) {
     enabled: false,
     mark_projection: 'inherit',
     mark_exclude_names: ['llm.chunk'],
+    attribute_mappings: [],
     transport: 'http_binary',
     headers: {},
     resource_attributes: {},
diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs
index adfaec0de..e4cf8d982 100644
--- a/crates/node/src/api/mod.rs
+++ b/crates/node/src/api/mod.rs
@@ -128,6 +128,22 @@ fn parse_string_map(
     Ok(out)
 }
 
+fn parse_attribute_mappings(
+    value: Option<Vec<OtlpAttributeMapping>>,
+) -> napi::Result<Vec<nemo_relay::observability::OtlpAttributeMapping>> {
+    let mappings = value
+        .unwrap_or_default()
+        .into_iter()
+        .map(|mapping| nemo_relay::observability::OtlpAttributeMapping {
+            key: mapping.key,
+            alias: mapping.alias,
+        })
+        .collect::<Vec<_>>();
+    nemo_relay::observability::validate_attribute_mappings(&mappings)
+        .map_err(napi::Error::from_reason)?;
+    Ok(mappings)
+}
+
 fn otel_status_metadata(status_code: &'static str, status_message: Option<String>) -> Json {
     let mut metadata = serde_json::Map::new();
     metadata.insert(
@@ -187,6 +203,7 @@ fn build_otel_config(
     for (key, value) in parse_string_map(options.resource_attributes, "resourceAttributes")? {
         config = config.with_resource_attribute(key, value);
     }
+    config = config.with_attribute_mappings(parse_attribute_mappings(options.attribute_mappings)?);
 
     Ok(config)
 }
@@ -304,6 +321,7 @@ fn build_openinference_config(
     for (key, value) in parse_string_map(options.resource_attributes, "resourceAttributes")? {
         config = config.with_resource_attribute(key, value);
     }
+    config = config.with_attribute_mappings(parse_attribute_mappings(options.attribute_mappings)?);
 
     Ok(config)
 }
@@ -3474,6 +3492,17 @@ pub struct OpenTelemetryConfig {
     pub instrumentation_scope: Option<String>,
     /// Export timeout in milliseconds. Defaults to `3000`.
     pub timeout_millis: Option<u32>,
+    /// Typed projected attributes copied to aliases.
+    pub attribute_mappings: Option<Vec<OtlpAttributeMapping>>,
+}
+
+/// Typed projected attribute copy configuration.
+#[napi(object)]
+pub struct OtlpAttributeMapping {
+    /// Fully-qualified projected attribute to copy.
+    pub key: String,
+    /// Additional attribute name receiving the copied value.
+    pub alias: String,
 }
 
 /// Mutable configuration object for `OpenInferenceSubscriber`.
@@ -3498,6 +3527,8 @@ pub struct OpenInferenceConfig {
     pub instrumentation_scope: Option<String>,
     /// Export timeout in milliseconds. Defaults to `3000`.
     pub timeout_millis: Option<u32>,
+    /// Typed projected attributes copied to aliases.
+    pub attribute_mappings: Option<Vec<OtlpAttributeMapping>>,
 }
 
 /// OpenTelemetry-backed event subscriber.
diff --git a/crates/node/tests/observability_plugin_tests.mjs b/crates/node/tests/observability_plugin_tests.mjs
index dd4c63d71..fb94b9982 100644
--- a/crates/node/tests/observability_plugin_tests.mjs
+++ b/crates/node/tests/observability_plugin_tests.mjs
@@ -31,6 +31,7 @@ describe('observability plugin helpers', () => {
       enabled: false,
       mark_projection: 'inherit',
       mark_exclude_names: ['llm.chunk'],
+      attribute_mappings: [],
       transport: 'http_binary',
       headers: {},
       resource_attributes: {},
diff --git a/crates/node/tests/openinference_tests.mjs b/crates/node/tests/openinference_tests.mjs
index 8ab01a977..a5f3d7f11 100644
--- a/crates/node/tests/openinference_tests.mjs
+++ b/crates/node/tests/openinference_tests.mjs
@@ -4,7 +4,7 @@
 import { describe, it } from 'node:test';
 import assert from 'node:assert/strict';
 import { createRequire } from 'node:module';
-import { startCollector } from '../../../scripts/test-support/otel_test_utils.mjs';
+import { assertOtlpStringAttribute, startCollector } from '../../../scripts/test-support/otel_test_utils.mjs';
 
 const require = createRequire(import.meta.url);
 const { OpenInferenceSubscriber, ScopeType, pushScope, popScope, event } = require('../index.js');
@@ -32,6 +32,7 @@ describe('OpenInferenceSubscriber', () => {
       resourceAttributes: {
         'deployment.environment': 'test',
       },
+      attributeMappings: [{ key: 'openinference.metadata.tenant', alias: 'tenant.id' }],
     });
 
     const name = unique('node_openinference');
@@ -68,6 +69,13 @@ describe('OpenInferenceSubscriber', () => {
         }),
       /resourceAttributes must be an object of string values/i,
     );
+    assert.throws(
+      () =>
+        new OpenInferenceSubscriber({
+          attributeMappings: [{ key: '', alias: 'tenant.id' }],
+        }),
+      /attribute mapping key must not be blank/i,
+    );
   });
 
   it('exports scope push/pop and mark events end to end', async () => {
@@ -75,21 +83,13 @@ describe('OpenInferenceSubscriber', () => {
     const subscriber = new OpenInferenceSubscriber({
       endpoint: collector.endpoint,
       serviceName: 'node-agent',
+      attributeMappings: [{ key: 'openinference.metadata.tenant', alias: 'tenant.id' }],
     });
 
     const name = unique('node_openinference_e2e');
     subscriber.register(name);
     try {
-      const scope = pushScope(
-        'openinference_scope',
-        ScopeType.Agent,
-        null,
-        null,
-        {
-          scope: true,
-        },
-        null,
-      );
+      const scope = pushScope('openinference_scope', ScopeType.Agent, null, null, null, null);
       event(
         'openinference_mark',
         scope,
@@ -100,7 +100,7 @@ describe('OpenInferenceSubscriber', () => {
           source: 'node',
         },
       );
-      popScope(scope);
+      popScope(scope, null, null, { tenant: 'node' });
 
       subscriber.forceFlush();
       const request = await collector.nextRequest();
@@ -111,6 +111,7 @@ describe('OpenInferenceSubscriber', () => {
       assertBodyContains(request.body, 'AGENT');
       assertBodyContains(request.body, 'metadata');
       assertBodyContains(request.body, 'openinference_mark');
+      assertOtlpStringAttribute(request.body, 'tenant.id', 'node');
     } finally {
       subscriber.deregister(name);
       subscriber.shutdown();
diff --git a/crates/node/tests/otel_tests.mjs b/crates/node/tests/otel_tests.mjs
index f9228d843..9144178b6 100644
--- a/crates/node/tests/otel_tests.mjs
+++ b/crates/node/tests/otel_tests.mjs
@@ -4,7 +4,7 @@
 import { describe, it } from 'node:test';
 import assert from 'node:assert/strict';
 import { createRequire } from 'node:module';
-import { startCollector } from '../../../scripts/test-support/otel_test_utils.mjs';
+import { assertOtlpStringAttribute, startCollector } from '../../../scripts/test-support/otel_test_utils.mjs';
 
 const require = createRequire(import.meta.url);
 const { OpenTelemetrySubscriber, ScopeType, pushScope, popScope, event } = require('../index.js');
@@ -13,6 +13,10 @@ function uniqueId(prefix) {
   return `${prefix}_${Date.now()}_${Math.random().toString(16).slice(2)}`;
 }
 
+function assertBodyContains(body, text) {
+  assert.equal(body.includes(Buffer.from(text, 'utf8')), true, `expected OTLP payload to contain ${text}`);
+}
+
 describe('OpenTelemetrySubscriber', () => {
   it('constructs from a mutable config object and supports lifecycle methods', () => {
     const subscriber = new OpenTelemetrySubscriber({
@@ -28,6 +32,7 @@ describe('OpenTelemetrySubscriber', () => {
       resourceAttributes: {
         'deployment.environment': 'test',
       },
+      attributeMappings: [{ key: 'nemo_relay.start.data.tenant', alias: 'tenant.id' }],
     });
 
     const name = uniqueId('node_otel');
@@ -64,6 +69,13 @@ describe('OpenTelemetrySubscriber', () => {
         }),
       /resourceAttributes must be an object of string values/i,
     );
+    assert.throws(
+      () =>
+        new OpenTelemetrySubscriber({
+          attributeMappings: [{ key: '', alias: 'tenant.id' }],
+        }),
+      /attribute mapping key must not be blank/i,
+    );
   });
 
   it('exports scope push/pop and mark events end to end', async () => {
@@ -71,21 +83,13 @@ describe('OpenTelemetrySubscriber', () => {
     const subscriber = new OpenTelemetrySubscriber({
       endpoint: collector.endpoint,
       serviceName: 'node-agent',
+      attributeMappings: [{ key: 'nemo_relay.mark.metadata.source', alias: 'tenant.id' }],
     });
 
     const name = uniqueId('node_otel_e2e');
     subscriber.register(name);
     try {
-      const scope = pushScope(
-        'otel_scope',
-        ScopeType.Agent,
-        null,
-        null,
-        {
-          scope: true,
-        },
-        null,
-      );
+      const scope = pushScope('otel_scope', ScopeType.Agent, null, null, null, null);
       event(
         'otel_mark',
         scope,
@@ -103,6 +107,7 @@ describe('OpenTelemetrySubscriber', () => {
       assert.equal(request.url, '/v1/traces');
       assert.equal(request.headers['content-type'], 'application/x-protobuf');
       assert.ok(request.body.length > 0);
+      assertOtlpStringAttribute(request.body, 'tenant.id', 'node');
     } finally {
       subscriber.deregister(name);
       subscriber.shutdown();
diff --git a/crates/python/src/py_types/observability.rs b/crates/python/src/py_types/observability.rs
index f51ed5990..85a24bb23 100644
--- a/crates/python/src/py_types/observability.rs
+++ b/crates/python/src/py_types/observability.rs
@@ -14,6 +14,21 @@ use super::{
     FORCE_ATIF_EXPORT_JSON_SERIALIZATION_ERROR, FORCE_ATIF_EXPORT_VALUE_SERIALIZATION_ERROR,
 };
 
+fn py_attribute_mappings(
+    value: &Bound<'_, PyAny>,
+) -> PyResult<Vec<nemo_relay::observability::OtlpAttributeMapping>> {
+    let value = py_to_json(value)?;
+    let mappings: Vec<nemo_relay::observability::OtlpAttributeMapping> =
+        serde_json::from_value(value).map_err(|error| {
+            pyo3::exceptions::PyValueError::new_err(format!(
+                "attribute_mappings must be a list of {{key: str, alias: str}} objects: {error}"
+            ))
+        })?;
+    nemo_relay::observability::validate_attribute_mappings(&mappings)
+        .map_err(pyo3::exceptions::PyValueError::new_err)?;
+    Ok(mappings)
+}
+
 // ---------------------------------------------------------------------------
 // AtifExporter
 // ---------------------------------------------------------------------------
@@ -427,6 +442,7 @@ pub struct PyOpenTelemetryConfig {
     pub(crate) timeout_millis: u64,
     pub(crate) headers: HashMap<String, String>,
     pub(crate) resource_attributes: HashMap<String, String>,
+    pub(crate) attribute_mappings: Vec<nemo_relay::observability::OtlpAttributeMapping>,
 }
 
 impl PyOpenTelemetryConfig {
@@ -464,6 +480,7 @@ impl PyOpenTelemetryConfig {
         for (key, value) in &self.resource_attributes {
             config = config.with_resource_attribute(key.clone(), value.clone());
         }
+        config = config.with_attribute_mappings(self.attribute_mappings.clone());
         Ok(config)
     }
 }
@@ -482,6 +499,7 @@ impl PyOpenTelemetryConfig {
             timeout_millis: 3_000,
             headers: HashMap::new(),
             resource_attributes: HashMap::new(),
+            attribute_mappings: Vec::new(),
         }
     }
 
@@ -513,6 +531,23 @@ impl PyOpenTelemetryConfig {
         Ok(())
     }
 
+    #[getter]
+    pub(crate) fn attribute_mappings(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
+        json_to_py(
+            py,
+            &serde_json::to_value(&self.attribute_mappings).unwrap_or_default(),
+        )
+    }
+
+    #[setter]
+    pub(crate) fn set_attribute_mappings(
+        &mut self,
+        attribute_mappings: &Bound<'_, PyAny>,
+    ) -> PyResult<()> {
+        self.attribute_mappings = py_attribute_mappings(attribute_mappings)?;
+        Ok(())
+    }
+
     pub(crate) fn set_header(&mut self, key: String, value: String) {
         self.headers.insert(key, value);
     }
@@ -606,6 +641,7 @@ pub struct PyOpenInferenceConfig {
     pub(crate) timeout_millis: u64,
     pub(crate) headers: HashMap<String, String>,
     pub(crate) resource_attributes: HashMap<String, String>,
+    pub(crate) attribute_mappings: Vec<nemo_relay::observability::OtlpAttributeMapping>,
 }
 
 impl PyOpenInferenceConfig {
@@ -643,6 +679,7 @@ impl PyOpenInferenceConfig {
         for (key, value) in &self.resource_attributes {
             config = config.with_resource_attribute(key.clone(), value.clone());
         }
+        config = config.with_attribute_mappings(self.attribute_mappings.clone());
         Ok(config)
     }
 }
@@ -661,6 +698,7 @@ impl PyOpenInferenceConfig {
             timeout_millis: 3_000,
             headers: HashMap::new(),
             resource_attributes: HashMap::new(),
+            attribute_mappings: Vec::new(),
         }
     }
 
@@ -692,6 +730,23 @@ impl PyOpenInferenceConfig {
         Ok(())
     }
 
+    #[getter]
+    pub(crate) fn attribute_mappings(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
+        json_to_py(
+            py,
+            &serde_json::to_value(&self.attribute_mappings).unwrap_or_default(),
+        )
+    }
+
+    #[setter]
+    pub(crate) fn set_attribute_mappings(
+        &mut self,
+        attribute_mappings: &Bound<'_, PyAny>,
+    ) -> PyResult<()> {
+        self.attribute_mappings = py_attribute_mappings(attribute_mappings)?;
+        Ok(())
+    }
+
     pub(crate) fn set_header(&mut self, key: String, value: String) {
         self.headers.insert(key, value);
     }
diff --git a/go/nemo_relay/nemo_relay.go b/go/nemo_relay/nemo_relay.go
index b4b884ec2..1f3ff2000 100644
--- a/go/nemo_relay/nemo_relay.go
+++ b/go/nemo_relay/nemo_relay.go
@@ -251,6 +251,7 @@ extern void nemo_relay_atof_exporter_free(void*);
 
 // OpenTelemetry subscriber
 extern int32_t nemo_relay_otel_subscriber_create(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, void**);
+extern int32_t nemo_relay_otel_subscriber_create_with_attribute_mappings(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, void**);
 extern int32_t nemo_relay_otel_subscriber_register(const void*, const char*);
 extern int32_t nemo_relay_otel_subscriber_deregister(const char*);
 extern int32_t nemo_relay_otel_subscriber_force_flush(const void*);
@@ -259,6 +260,7 @@ extern void nemo_relay_otel_subscriber_free(void*);
 
 // OpenInference subscriber
 extern int32_t nemo_relay_openinference_subscriber_create(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, void**);
+extern int32_t nemo_relay_openinference_subscriber_create_with_attribute_mappings(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, void**);
 extern int32_t nemo_relay_openinference_subscriber_register(const void*, const char*);
 extern int32_t nemo_relay_openinference_subscriber_deregister(const char*);
 extern int32_t nemo_relay_openinference_subscriber_force_flush(const void*);
@@ -1868,6 +1870,13 @@ type OpenTelemetryConfig struct {
 	ServiceVersion       string
 	InstrumentationScope string
 	Timeout              time.Duration
+	AttributeMappings    []OtlpAttributeMapping
+}
+
+// OtlpAttributeMapping copies a projected OTLP attribute to an alias.
+type OtlpAttributeMapping struct {
+	Key   string `json:"key"`
+	Alias string `json:"alias"`
 }
 
 // NewOpenTelemetryConfig returns a config initialized with sensible defaults.
@@ -1931,6 +1940,16 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc
 	cResourceAttrsJSON := C.CString(string(resourceAttrsJSON))
 	defer C.free(unsafe.Pointer(cResourceAttrsJSON))
 
+	var cAttributeMappingsJSON *C.char
+	if config.AttributeMappings != nil {
+		attributeMappingsJSON, err := jsonMarshal(config.AttributeMappings)
+		if err != nil {
+			return nil, err
+		}
+		cAttributeMappingsJSON = C.CString(string(attributeMappingsJSON))
+		defer C.free(unsafe.Pointer(cAttributeMappingsJSON))
+	}
+
 	cServiceName := C.CString(config.ServiceName)
 	defer C.free(unsafe.Pointer(cServiceName))
 
@@ -1950,7 +1969,7 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc
 	defer C.free(unsafe.Pointer(cInstrumentationScope))
 
 	var ptr unsafe.Pointer
-	status := C.nemo_relay_otel_subscriber_create(
+	status := C.nemo_relay_otel_subscriber_create_with_attribute_mappings(
 		cTransport,
 		cEndpoint,
 		cHeadersJSON,
@@ -1960,6 +1979,7 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc
 		cServiceVersion,
 		cInstrumentationScope,
 		C.uint64_t(config.Timeout/time.Millisecond),
+		cAttributeMappingsJSON,
 		&ptr,
 	)
 	if err := checkStatus(status); err != nil {
@@ -2032,6 +2052,7 @@ type OpenInferenceConfig struct {
 	ServiceVersion       string
 	InstrumentationScope string
 	Timeout              time.Duration
+	AttributeMappings    []OtlpAttributeMapping
 }
 
 // NewOpenInferenceConfig returns a config initialized with sensible defaults.
@@ -2095,6 +2116,16 @@ func NewOpenInferenceSubscriber(config OpenInferenceConfig) (*OpenInferenceSubsc
 	cResourceAttrsJSON := C.CString(string(resourceAttrsJSON))
 	defer C.free(unsafe.Pointer(cResourceAttrsJSON))
 
+	var cAttributeMappingsJSON *C.char
+	if config.AttributeMappings != nil {
+		attributeMappingsJSON, err := jsonMarshal(config.AttributeMappings)
+		if err != nil {
+			return nil, err
+		}
+		cAttributeMappingsJSON = C.CString(string(attributeMappingsJSON))
+		defer C.free(unsafe.Pointer(cAttributeMappingsJSON))
+	}
+
 	cServiceName := C.CString(config.ServiceName)
 	defer C.free(unsafe.Pointer(cServiceName))
 
@@ -2114,7 +2145,7 @@ func NewOpenInferenceSubscriber(config OpenInferenceConfig) (*OpenInferenceSubsc
 	defer C.free(unsafe.Pointer(cInstrumentationScope))
 
 	var ptr unsafe.Pointer
-	status := C.nemo_relay_openinference_subscriber_create(
+	status := C.nemo_relay_openinference_subscriber_create_with_attribute_mappings(
 		cTransport,
 		cEndpoint,
 		cHeadersJSON,
@@ -2124,6 +2155,7 @@ func NewOpenInferenceSubscriber(config OpenInferenceConfig) (*OpenInferenceSubsc
 		cServiceVersion,
 		cInstrumentationScope,
 		C.uint64_t(config.Timeout/time.Millisecond),
+		cAttributeMappingsJSON,
 		&ptr,
 	)
 	if err := checkStatus(status); err != nil {
diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go
index b25baee77..27047596e 100644
--- a/go/nemo_relay/observability_plugin.go
+++ b/go/nemo_relay/observability_plugin.go
@@ -184,6 +184,7 @@ type ObservabilityOtlpConfig struct {
 	Enabled              bool                        `json:"enabled,omitempty"`
 	MarkProjection       ObservabilityMarkProjection `json:"mark_projection,omitempty"`
 	MarkExcludeNames     []string                    `json:"mark_exclude_names,omitempty"`
+	AttributeMappings    []OtlpAttributeMapping      `json:"attribute_mappings,omitempty"`
 	Transport            string                      `json:"transport,omitempty"`
 	Endpoint             string                      `json:"endpoint,omitempty"`
 	Headers              map[string]string           `json:"headers,omitempty"`
diff --git a/go/nemo_relay/openinference_test.go b/go/nemo_relay/openinference_test.go
index 75f5f8fb9..0f3c8c36f 100644
--- a/go/nemo_relay/openinference_test.go
+++ b/go/nemo_relay/openinference_test.go
@@ -52,6 +52,10 @@ func TestOpenInferenceSubscriberLifecycle(t *testing.T) {
 	config.Timeout = 1250 * time.Millisecond
 	config.Headers["authorization"] = "Bearer token"
 	config.ResourceAttributes["deployment.environment"] = "test"
+	config.AttributeMappings = []OtlpAttributeMapping{{
+		Key:   "openinference.metadata.tenant",
+		Alias: "tenant.id",
+	}}
 
 	subscriber, err := NewOpenInferenceSubscriber(config)
 	if err != nil {
@@ -87,14 +91,55 @@ func TestOpenInferenceSubscriberRejectsInvalidTransport(t *testing.T) {
 	}
 }
 
-func TestOpenInferenceSubscriberExportsScopeLifecycleAndMarks(t *testing.T) {
+func TestOpenInferenceSubscriberRejectsInvalidAttributeMapping(t *testing.T) {
+	config := NewOpenInferenceConfig()
+	config.AttributeMappings = []OtlpAttributeMapping{{Key: "", Alias: "tenant.id"}}
+
+	if _, err := NewOpenInferenceSubscriber(config); err == nil {
+		t.Fatal("expected invalid attribute mapping error")
+	}
+}
+
+func TestOpenInferenceSubscriberExportsScopeLifecycleAndMappedAttributes(t *testing.T) {
 	requests := make(chan otelRequest, 4)
 	server := NewOtelTestServer(t, requests)
 	defer server.Close()
 
-	subscriber := NewRegisteredOpenInferenceSubscriber(t, server.URL+"/v1/traces")
+	config := NewOpenInferenceConfig()
+	config.Endpoint = server.URL + "/v1/traces"
+	config.ServiceName = "go-agent"
+	config.AttributeMappings = []OtlpAttributeMapping{{
+		Key:   "openinference.metadata.tenant",
+		Alias: "tenant.id",
+	}}
+	subscriber, err := NewOpenInferenceSubscriber(config)
+	if err != nil {
+		t.Fatalf("NewOpenInferenceSubscriber failed: %v", err)
+	}
 	defer subscriber.Close()
-	EmitOpenInferenceScopeLifecycle(t)
+	name := "go_openinference_e2e_" + time.Now().Format("150405.000000")
+	if err := subscriber.Register(name); err != nil {
+		t.Fatalf("Register failed: %v", err)
+	}
+	defer func() { _ = subscriber.Deregister(name) }()
+
+	runWithTestScopeStack(t, func() {
+		handle, err := PushScope("openinference_scope", ScopeTypeAgent)
+		if err != nil {
+			t.Fatalf("PushScope failed: %v", err)
+		}
+		requireNoError(t, EmitEvent(
+			"openinference_mark",
+			WithEventParent(handle),
+			WithEventData(json.RawMessage(`{"step":1}`)),
+			WithEventMetadata(json.RawMessage(`{"source":"go"}`)),
+		), "EmitEvent failed")
+		requireNoError(
+			t,
+			PopScope(handle, WithScopeEndMetadata(json.RawMessage(`{"tenant":"go"}`))),
+			"PopScope failed",
+		)
+	})
 	if err := subscriber.ForceFlush(); err != nil {
 		t.Fatalf("ForceFlush failed: %v", err)
 	}
@@ -102,6 +147,7 @@ func TestOpenInferenceSubscriberExportsScopeLifecycleAndMarks(t *testing.T) {
 	select {
 	case request := <-requests:
 		AssertOpenInferenceRequest(t, request)
+		assertOtlpStringAttribute(t, request.Body, "tenant.id", "go")
 	case <-time.After(5 * time.Second):
 		t.Fatal("timed out waiting for OTLP request")
 	}
diff --git a/go/nemo_relay/otel_test.go b/go/nemo_relay/otel_test.go
index 4ba6d7870..73850d591 100644
--- a/go/nemo_relay/otel_test.go
+++ b/go/nemo_relay/otel_test.go
@@ -4,6 +4,8 @@
 package nemo_relay
 
 import (
+	"bytes"
+	"encoding/binary"
 	"encoding/json"
 	"io"
 	"net/http"
@@ -12,6 +14,20 @@ import (
 	"time"
 )
 
+func assertOtlpStringAttribute(t *testing.T, body []byte, key string, value string) {
+	t.Helper()
+	encoded := append([]byte{0x0a}, binary.AppendUvarint(nil, uint64(len(key)))...)
+	encoded = append(encoded, key...)
+	attributeValue := append([]byte{0x0a}, binary.AppendUvarint(nil, uint64(len(value)))...)
+	attributeValue = append(attributeValue, value...)
+	encoded = append(encoded, 0x12)
+	encoded = binary.AppendUvarint(encoded, uint64(len(attributeValue)))
+	encoded = append(encoded, attributeValue...)
+	if !bytes.Contains(body, encoded) {
+		t.Fatalf("expected OTLP string attribute %s=%s", key, value)
+	}
+}
+
 func TestNewOpenTelemetryConfigDefaults(t *testing.T) {
 	config := NewOpenTelemetryConfig()
 
@@ -45,6 +61,10 @@ func TestOpenTelemetrySubscriberLifecycle(t *testing.T) {
 	config.Timeout = 1250 * time.Millisecond
 	config.Headers["authorization"] = "Bearer token"
 	config.ResourceAttributes["deployment.environment"] = "test"
+	config.AttributeMappings = []OtlpAttributeMapping{{
+		Key:   "nemo_relay.start.data.tenant",
+		Alias: "tenant.id",
+	}}
 
 	subscriber, err := NewOpenTelemetrySubscriber(config)
 	if err != nil {
@@ -80,6 +100,15 @@ func TestOpenTelemetrySubscriberRejectsInvalidTransport(t *testing.T) {
 	}
 }
 
+func TestOpenTelemetrySubscriberRejectsInvalidAttributeMapping(t *testing.T) {
+	config := NewOpenTelemetryConfig()
+	config.AttributeMappings = []OtlpAttributeMapping{{Key: "", Alias: "tenant.id"}}
+
+	if _, err := NewOpenTelemetrySubscriber(config); err == nil {
+		t.Fatal("expected invalid attribute mapping error")
+	}
+}
+
 func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) {
 	type otelRequest struct {
 		Path        string
@@ -105,6 +134,10 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) {
 	config := NewOpenTelemetryConfig()
 	config.Endpoint = server.URL + "/v1/traces"
 	config.ServiceName = "go-agent"
+	config.AttributeMappings = []OtlpAttributeMapping{{
+		Key:   "nemo_relay.mark.metadata.source",
+		Alias: "tenant.id",
+	}}
 
 	subscriber, err := NewOpenTelemetrySubscriber(config)
 	if err != nil {
@@ -150,6 +183,7 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) {
 		if len(request.Body) == 0 {
 			t.Fatal("expected non-empty OTLP request body")
 		}
+		assertOtlpStringAttribute(t, request.Body, "tenant.id", "go")
 	case <-time.After(5 * time.Second):
 		t.Fatal("timed out waiting for OTLP request")
 	}
diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi
index b9e551304..d540da1cf 100644
--- a/python/nemo_relay/_native.pyi
+++ b/python/nemo_relay/_native.pyi
@@ -972,6 +972,14 @@ class OpenTelemetryConfig:
     def resource_attributes(self, value: dict[str, str]) -> None:
         """Replace additional OpenTelemetry resource attributes."""
         ...
+    @property
+    def attribute_mappings(self) -> list[dict[str, str]]:
+        """Return typed projected-attribute aliases."""
+        ...
+    @attribute_mappings.setter
+    def attribute_mappings(self, value: list[dict[str, str]]) -> None:
+        """Replace typed projected-attribute aliases."""
+        ...
     def set_header(self, key: str, value: str) -> None:
         """Set one exporter header key/value pair."""
         ...
@@ -1044,6 +1052,14 @@ class OpenInferenceConfig:
     def resource_attributes(self, value: dict[str, str]) -> None:
         """Replace additional OpenInference resource attributes."""
         ...
+    @property
+    def attribute_mappings(self) -> list[dict[str, str]]:
+        """Return typed projected-attribute aliases."""
+        ...
+    @attribute_mappings.setter
+    def attribute_mappings(self, value: list[dict[str, str]]) -> None:
+        """Replace typed projected-attribute aliases."""
+        ...
     def set_header(self, key: str, value: str) -> None:
         """Set one exporter header key/value pair."""
         ...
diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py
index 1495ba9f7..6dde6a9b2 100644
--- a/python/nemo_relay/observability.py
+++ b/python/nemo_relay/observability.py
@@ -227,6 +227,7 @@ class OtlpConfig:
     service_version: str | None = None
     instrumentation_scope: str | None = None
     timeout_millis: int = 3000
+    attribute_mappings: list[dict[str, str]] = field(default_factory=list)
 
     def to_dict(self) -> JsonObject:
         """Serialize this OTLP config to the canonical JSON object shape."""
@@ -235,6 +236,7 @@ def to_dict(self) -> JsonObject:
                 "enabled": self.enabled,
                 "mark_projection": self.mark_projection,
                 "mark_exclude_names": self.mark_exclude_names,
+                "attribute_mappings": self.attribute_mappings,
                 "transport": self.transport,
                 "endpoint": self.endpoint,
                 "headers": self.headers,
diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi
index bdeaf236b..d8333c020 100644
--- a/python/nemo_relay/observability.pyi
+++ b/python/nemo_relay/observability.pyi
@@ -91,6 +91,7 @@ class OtlpConfig:
     service_version: str | None = ...
     instrumentation_scope: str | None = ...
     timeout_millis: int = ...
+    attribute_mappings: list[dict[str, str]] = field(default_factory=list)
     def to_dict(self) -> JsonObject: ...
 
 @dataclass(slots=True)
diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py
index 7839f16e3..4c1336e55 100644
--- a/python/tests/test_observability_plugin.py
+++ b/python/tests/test_observability_plugin.py
@@ -41,6 +41,7 @@ def test_defaults_and_component_wrapper(self):
             "enabled": False,
             "mark_projection": "inherit",
             "mark_exclude_names": ["llm.chunk"],
+            "attribute_mappings": [],
             "transport": "http_binary",
             "headers": {},
             "resource_attributes": {},
diff --git a/python/tests/test_types.py b/python/tests/test_types.py
index 9ef4a45b5..f537216b1 100644
--- a/python/tests/test_types.py
+++ b/python/tests/test_types.py
@@ -91,6 +91,22 @@ class _OtelCollectorServer(http.server.ThreadingHTTPServer):
     request_event: threading.Event
 
 
+def _encode_varint(value: int) -> bytes:
+    result = bytearray()
+    while value >= 0x80:
+        result.append((value & 0x7F) | 0x80)
+        value >>= 7
+    result.append(value)
+    return bytes(result)
+
+
+def _otlp_string_attribute(key: str, value: str) -> bytes:
+    key_bytes = key.encode()
+    value_bytes = value.encode()
+    any_value = b"\x0a" + _encode_varint(len(value_bytes)) + value_bytes
+    return b"\x0a" + _encode_varint(len(key_bytes)) + key_bytes + b"\x12" + _encode_varint(len(any_value)) + any_value
+
+
 def _scope_event(events, name: str, category: str, scope_category: str) -> ScopeEvent:
     return next(
         event
@@ -572,6 +588,8 @@ def test_config_defaults_mutation_and_repr(self):
 
         assert config.headers == {"authorization": "Bearer token"}
         assert config.resource_attributes == {"deployment.environment": "test"}
+        config.attribute_mappings = [{"key": "nemo_relay.start.metadata.tenant", "alias": "tenant.id"}]
+        assert config.attribute_mappings == [{"key": "nemo_relay.start.metadata.tenant", "alias": "tenant.id"}]
         assert "OpenTelemetryConfig" in repr(config)
 
     def test_config_rejects_invalid_map_values(self):
@@ -583,6 +601,18 @@ def test_config_rejects_invalid_map_values(self):
         with pytest.raises(ValueError, match="dict\\[str, str\\]"):
             config.resource_attributes = cast(dict[str, str], {"env": 1})
 
+        with pytest.raises(ValueError, match="attribute mapping key must not be blank"):
+            config.attribute_mappings = [{"key": "", "alias": "tenant.id"}]
+
+        with pytest.raises(ValueError, match="attribute mapping alias must not be blank"):
+            config.attribute_mappings = [{"key": "nemo_relay.mark.metadata.source", "alias": ""}]
+
+        with pytest.raises(ValueError, match=r"attribute mapping alias .* duplicated"):
+            config.attribute_mappings = [
+                {"key": "one", "alias": "tenant.id"},
+                {"key": "two", "alias": "tenant.id"},
+            ]
+
     def test_subscriber_lifecycle_and_invalid_transport(self):
         config = OpenTelemetryConfig()
         config.endpoint = "http://localhost:4318/v1/traces"
@@ -608,9 +638,11 @@ def test_subscriber_lifecycle_and_invalid_transport(self):
 
     def test_subscriber_exports_scope_and_mark_events_end_to_end(self):
         with _OtelCollector() as collector:
+            source = "python-é" * 20
             config = OpenTelemetryConfig()
             config.endpoint = collector.endpoint
             config.service_name = "py-agent"
+            config.attribute_mappings = [{"key": "nemo_relay.mark.metadata.source", "alias": "tenant.id"}]
 
             subscriber = OpenTelemetrySubscriber(config)
             subscriber_name = f"py_otel_e2e_{uuid4().hex}"
@@ -623,7 +655,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self):
                         "otel_mark",
                         handle=handle,
                         data={"step": 1},
-                        metadata={"source": "python"},
+                        metadata={"source": source},
                     )
                 finally:
                     scope.pop(handle)
@@ -633,6 +665,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self):
                 assert request["path"] == "/v1/traces"
                 assert request["headers"]["content-type"] == "application/x-protobuf"
                 assert request["body"]
+                assert _otlp_string_attribute("tenant.id", source) in request["body"]
             finally:
                 subscriber.deregister(subscriber_name)
                 subscriber.shutdown()
@@ -661,6 +694,8 @@ def test_config_defaults_mutation_and_repr(self):
 
         assert config.headers == {"authorization": "Bearer token"}
         assert config.resource_attributes == {"deployment.environment": "test"}
+        config.attribute_mappings = [{"key": "openinference.metadata.tenant", "alias": "tenant.id"}]
+        assert config.attribute_mappings == [{"key": "openinference.metadata.tenant", "alias": "tenant.id"}]
         assert "OpenInferenceConfig" in repr(config)
 
     def test_config_rejects_invalid_map_values(self):
@@ -672,6 +707,18 @@ def test_config_rejects_invalid_map_values(self):
         with pytest.raises(ValueError, match="dict\\[str, str\\]"):
             config.resource_attributes = cast(dict[str, str], {"env": 1})
 
+        with pytest.raises(ValueError, match="attribute mapping key must not be blank"):
+            config.attribute_mappings = [{"key": "", "alias": "tenant.id"}]
+
+        with pytest.raises(ValueError, match="attribute mapping alias must not be blank"):
+            config.attribute_mappings = [{"key": "openinference.metadata.tenant", "alias": ""}]
+
+        with pytest.raises(ValueError, match=r"attribute mapping alias .* duplicated"):
+            config.attribute_mappings = [
+                {"key": "one", "alias": "tenant.id"},
+                {"key": "two", "alias": "tenant.id"},
+            ]
+
     def test_subscriber_lifecycle_and_invalid_transport(self):
         config = OpenInferenceConfig()
         config.endpoint = "http://localhost:4318/v1/traces"
@@ -704,9 +751,11 @@ def test_subscriber_lifecycle_and_invalid_transport(self):
 
     def test_subscriber_exports_scope_and_mark_events_end_to_end(self):
         with _OtelCollector() as collector:
+            source = "python-é" * 20
             config = OpenInferenceConfig()
             config.endpoint = collector.endpoint
             config.service_name = "py-agent"
+            config.attribute_mappings = [{"key": "nemo_relay.mark.metadata.source", "alias": "tenant.id"}]
 
             subscriber = OpenInferenceSubscriber(config)
             subscriber_name = f"py_openinference_e2e_{uuid4().hex}"
@@ -719,7 +768,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self):
                         "openinference_mark",
                         handle=handle,
                         data={"step": 1},
-                        metadata={"source": "python"},
+                        metadata={"source": source},
                     )
                 finally:
                     scope.pop(handle)
@@ -733,6 +782,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self):
                 assert b"AGENT" in request["body"]
                 assert b"metadata" in request["body"]
                 assert b"openinference_mark" in request["body"]
+                assert _otlp_string_attribute("tenant.id", source) in request["body"]
             finally:
                 subscriber.deregister(subscriber_name)
                 subscriber.shutdown()
diff --git a/scripts/test-support/otel_test_utils.mjs b/scripts/test-support/otel_test_utils.mjs
index e7fd87117..8e7727321 100644
--- a/scripts/test-support/otel_test_utils.mjs
+++ b/scripts/test-support/otel_test_utils.mjs
@@ -5,6 +5,33 @@ import { spawn } from 'node:child_process';
 import { once } from 'node:events';
 import { fileURLToPath } from 'node:url';
 
+function encodeVarint(value) {
+  const bytes = [];
+  do {
+    const byte = value % 128;
+    value = Math.floor(value / 128);
+    bytes.push(value === 0 ? byte : byte | 0x80);
+  } while (value !== 0);
+  return Buffer.from(bytes);
+}
+
+export function assertOtlpStringAttribute(body, key, value) {
+  const keyBuffer = Buffer.from(key, 'utf8');
+  const valueBuffer = Buffer.from(value, 'utf8');
+  const attributeValue = Buffer.concat([Buffer.from([0x0a]), encodeVarint(valueBuffer.length), valueBuffer]);
+  const encoded = Buffer.concat([
+    Buffer.from([0x0a]),
+    encodeVarint(keyBuffer.length),
+    keyBuffer,
+    Buffer.from([0x12]),
+    encodeVarint(attributeValue.length),
+    attributeValue,
+  ]);
+  if (!body.includes(encoded)) {
+    throw new Error(`expected OTLP string attribute ${key}=${value}`);
+  }
+}
+
 export async function startCollector() {
   const requests = [];
   let nextRequestIndex = 0;
@@ -68,7 +95,9 @@ export async function startCollector() {
   });
   const endpoint = await Promise.race([
     readyPromise,
-    new Promise((_, reject) => setTimeout(() => reject(new Error('timed out waiting for OTLP collector startup')), 5000)),
+    new Promise((_, reject) =>
+      setTimeout(() => reject(new Error('timed out waiting for OTLP collector startup')), 5000),
+    ),
   ]);
 
   return {
@@ -86,7 +115,9 @@ export async function startCollector() {
       });
       return await Promise.race([
         requestPromise,
-        new Promise((_, reject) => setTimeout(() => reject(new Error('timed out waiting for OTLP request')), timeoutMs)),
+        new Promise((_, reject) =>
+          setTimeout(() => reject(new Error('timed out waiting for OTLP request')), timeoutMs),
+        ),
       ]);
     },
     async close() {