chore: Set status on OTel spans

crates/core/src/api/llm.rs

@@ -19,8 +19,8 @@ use crate::api::runtime::{
 use crate::api::scope::event;
 use crate::api::scope::{EmitMarkEventParams, ScopeHandle};
 use crate::api::shared::{
-    ensure_runtime_owner, resolve_parent_uuid, run_request_intercepts_with_codec,
-    snapshot_event_subscribers,
+    ensure_runtime_owner, metadata_with_otel_status, resolve_parent_uuid,
+    run_request_intercepts_with_codec, snapshot_event_subscribers,
 };
 use crate::codec::request::AnnotatedLlmRequest;
 use crate::codec::response::AnnotatedLlmResponse;
@@ -587,19 +587,22 @@ pub async fn llm_call_execute(params: LlmCallExecuteParams) -> Result<Json> {
                 .as_ref()
                 .and_then(|codec| codec.decode_response(&response).ok())
                 .map(Arc::new);
+            let end_metadata = metadata_with_otel_status(metadata, "OK", None);
             llm_call_end(
                 LlmCallEndParams::builder()
                     .handle(&handle)
                     .response(response.clone())
                     .data_opt(data)
-                    .metadata_opt(metadata)
+                    .metadata_opt(end_metadata)
                     .annotated_response_opt(annotated_response)
                     .build(),
             )?;
             Ok(response)
         }
         Err(error) => {
-            let _ = emit_llm_end_without_output(&handle, metadata);
+            let end_metadata =
+                metadata_with_otel_status(metadata, "ERROR", Some(error.to_string()));
+            let _ = emit_llm_end_without_output(&handle, end_metadata);
             Err(error)
         }
     }
@@ -732,19 +735,22 @@ pub async fn llm_stream_call_execute(params: LlmStreamCallExecuteParams) -> Resu
 
     match execution(intercepted_request).await {
         Ok(raw_stream) => {
+            let end_metadata = metadata_with_otel_status(metadata, "OK", None);
             let wrapper = LlmStreamWrapper::new(
                 raw_stream,
                 handle,
                 collector,
                 finalizer,
                 data,
-                metadata,
+                end_metadata,
                 response_codec,
             );
             Ok(Box::pin(wrapper) as LlmJsonStream)
         }
         Err(error) => {
-            let _ = emit_llm_end_without_output(&handle, metadata);
+            let end_metadata =
+                metadata_with_otel_status(metadata, "ERROR", Some(error.to_string()));
+            let _ = emit_llm_end_without_output(&handle, end_metadata);
             Err(error)
         }
     }
@@ -831,3 +837,7 @@ pub fn llm_conditional_execution(request: &LlmRequest) -> Result<()> {
     }
     Ok(())
 }
+
+#[cfg(test)]
+#[path = "../../tests/unit/llm_api_tests.rs"]
+mod tests;

crates/core/src/api/runtime/state.rs

@@ -247,20 +247,30 @@ impl NemoRelayContextState {
     /// # Parameters
     /// - `handle`: Scope handle to serialize into an event.
     /// - `data`: Optional data payload returned from the scope.
+    /// - `metadata`: Optional metadata payload merged over `handle.metadata`.
     ///
     /// # Returns
     /// A scope-end [`Event`] derived from the provided handle.
-    pub fn end_scope_handle(&self, handle: &ScopeHandle, data: Option<Json>) -> Event {
+    pub fn end_scope_handle(
+        &self,
+        handle: &ScopeHandle,
+        data: Option<Json>,
+        metadata: Option<Json>,
+    ) -> Event {
         self.build_scope_end_event(
             EndScopeHandleParams::builder()
                 .handle(handle)
                 .data_opt(data)
+                .metadata_opt(metadata)
                 .build(),
         )
     }
 
     /// Build a scope-end event from builder parameters.
     ///
+    /// The `metadata` payload is merged over the metadata already stored on
+    /// the handle.
+    ///
     /// # Parameters
     /// - `params`: Scope end-event builder parameters.
     ///
@@ -279,7 +289,7 @@ impl NemoRelayContextState {
                 )
                 .name(handle.name.as_str())
                 .data_opt(params.data)
-                .metadata_opt(handle.metadata.clone())
+                .metadata_opt(merge_json(handle.metadata.clone(), params.metadata))
                 .build(),
             ScopeCategory::End,
             scope_attributes_to_strings(handle.attributes),

crates/core/src/api/scope.rs

@@ -180,6 +180,9 @@ pub struct EndScopeHandleParams<'a> {
     /// Optional JSON payload exported as the semantic scope output.
     #[builder(default)]
     pub data: Option<Json>,
+    /// Optional metadata to be appended to the metadata set when the scope was created.
+    #[builder(default)]
+    pub metadata: Option<Json>,
     /// Optional timestamp recorded on the emitted end event. When omitted, the
     /// runtime records the current UTC time, or one microsecond after the
     /// handle start time if the current time is not later.
@@ -196,6 +199,9 @@ pub struct PopScopeParams<'a> {
     /// Optional JSON payload exported as the semantic scope output.
     #[builder(default)]
     pub output: Option<Json>,
+    /// Optional JSON payload metadata to be appended to the metadata set when the scope was created.
+    #[builder(default)]
+    pub metadata: Option<Json>,
     /// Optional timestamp recorded on the emitted end event. When omitted, the
     /// runtime records the current UTC time, or one microsecond after the
     /// handle start time if the current time is not later.
@@ -347,6 +353,7 @@ pub fn pop_scope(params: PopScopeParams<'_>) -> Result<()> {
                 .handle(&scope)
                 .data_opt(params.output)
                 .timestamp_opt(params.timestamp)
+                .metadata_opt(params.metadata)
                 .build(),
         );
         (scope, event, subscribers)

crates/core/src/api/shared.rs

@@ -13,6 +13,7 @@ use crate::api::scope::ScopeHandle;
 use crate::codec::request::AnnotatedLlmRequest;
 use crate::codec::traits::LlmCodec;
 use crate::error::{FlowError, Result};
+use crate::json::{Json, merge_json};
 use crate::shared_runtime::ensure_process_runtime_owner;
 
 pub(crate) fn resolve_parent_uuid(parent: Option<&ScopeHandle>) -> Option<Uuid> {
@@ -37,6 +38,25 @@ pub(crate) fn ensure_runtime_owner() -> Result<()> {
     ensure_process_runtime_owner()
 }
 
+pub(crate) fn metadata_with_otel_status(
+    metadata: Option<Json>,
+    status_code: &'static str,
+    status_message: Option<String>,
+) -> Option<Json> {
+    let mut status = serde_json::Map::new();
+    status.insert(
+        "otel.status_code".to_string(),
+        Json::String(status_code.to_string()),
+    );
+    if let Some(status_message) = status_message {
+        status.insert(
+            "otel.status_message".to_string(),
+            Json::String(status_message),
+        );
+    }
+    merge_json(metadata, Some(Json::Object(status)))
+}
+
 pub(crate) fn run_request_intercepts_with_codec(
     name: &str,
     request: LlmRequest,

crates/core/src/api/tool.rs

@@ -9,7 +9,10 @@ use crate::api::runtime::current_scope_stack;
 use crate::api::runtime::global_context;
 use crate::api::scope::event;
 use crate::api::scope::{EmitMarkEventParams, ScopeHandle};
-use crate::api::shared::{ensure_runtime_owner, resolve_parent_uuid, snapshot_event_subscribers};
+use crate::api::shared::{
+    ensure_runtime_owner, metadata_with_otel_status, resolve_parent_uuid,
+    snapshot_event_subscribers,
+};
 use crate::error::{FlowError, Result};
 use crate::json::Json;
 use bitflags::bitflags;
@@ -445,18 +448,21 @@ pub async fn tool_call_execute(params: ToolCallExecuteParams) -> Result<Json> {
 
     match execution(intercepted_args).await {
         Ok(result) => {
+            let end_metadata = metadata_with_otel_status(metadata, "OK", None);
             tool_call_end(
                 ToolCallEndParams::builder()
                     .handle(&handle)
                     .result(result.clone())
                     .data_opt(data)
-                    .metadata_opt(metadata)
+                    .metadata_opt(end_metadata)
                     .build(),
             )?;
             Ok(result)
         }
         Err(error) => {
-            let _ = emit_tool_end_without_output(&handle, metadata);
+            let end_metadata =
+                metadata_with_otel_status(metadata, "ERROR", Some(error.to_string()));
+            let _ = emit_tool_end_without_output(&handle, end_metadata);
             Err(error)
         }
     }
@@ -542,3 +548,7 @@ pub fn tool_conditional_execution(name: &str, args: &Json) -> Result<()> {
     }
     Ok(())
 }
+
+#[cfg(test)]
+#[path = "../../tests/unit/tool_api_tests.rs"]
+mod tests;

crates/core/src/observability/mod.rs

@@ -18,3 +18,36 @@ pub mod openinference;
 #[cfg(feature = "otel")]
 pub mod otel;
 pub mod plugin_component;
+
+#[cfg(any(feature = "otel", feature = "openinference"))]
+pub(crate) fn set_span_status_from_event_metadata<S>(span: &mut S, event: &crate::api::event::Event)
+where
+    S: opentelemetry::trace::Span,
+{
+    let Some(metadata) = event.metadata() else {
+        return;
+    };
+    let Some(status_code) = metadata
+        .get("otel.status_code")
+        .and_then(crate::json::Json::as_str)
+    else {
+        return;
+    };
+
+    let status = match status_code {
+        "OK" => opentelemetry::trace::Status::Ok,
+        "ERROR" => opentelemetry::trace::Status::error(
+            metadata
+                .get("otel.status_message")
+                .and_then(crate::json::Json::as_str)
+                .unwrap_or_default()
+                .to_string(),
+        ),
+        "UNSET" => opentelemetry::trace::Status::Unset,
+        other => {
+            eprintln!("Unrecognized OTEL status code in event metadata: {other}");
+            opentelemetry::trace::Status::Unset
+        }
+    };
+    span.set_status(status);
+}

crates/core/src/observability/openinference.rs

@@ -549,6 +549,7 @@ impl OpenInferenceEventProcessor {
         let Some(mut active_span) = self.active_spans.remove(&event.uuid()) else {
             return;
         };
+        super::set_span_status_from_event_metadata(&mut active_span.span, event);
         active_span.span.set_attributes(end_attributes(event));
         active_span
             .span

crates/core/src/observability/otel.rs

@@ -543,6 +543,8 @@ impl OtelEventProcessor {
         let Some(mut active_span) = self.active_spans.remove(&event.uuid()) else {
             return;
         };
+
+        super::set_span_status_from_event_metadata(&mut active_span.span, event);
         active_span.span.set_attributes(end_attributes(event));
         active_span
             .span
@@ -657,16 +659,15 @@ fn start_attributes(event: &Event) -> Vec<KeyValue> {
 fn end_attributes(event: &Event) -> Vec<KeyValue> {
     let mut attributes = Vec::new();
     push_serialized(&mut attributes, "nemo_relay.end.data_json", event.data());
-    push_serialized(
-        &mut attributes,
-        "nemo_relay.end.metadata_json",
-        event.metadata(),
-    );
+
+    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(),
     );
+
     attributes
 }
 

crates/core/tests/unit/context_tests.rs

@@ -15,7 +15,7 @@ use crate::api::runtime::EventSubscriberFn;
 use crate::api::runtime::ScopeStack;
 use crate::api::runtime::global_context;
 use crate::api::runtime::{NemoRelayContextState, flush_subscribers};
-use crate::api::scope::{ScopeAttributes, ScopeHandle, ScopeType};
+use crate::api::scope::{EndScopeHandleParams, ScopeAttributes, ScopeHandle, ScopeType};
 use crate::api::tool::CreateToolHandleParams;
 use crate::context::registries::{
     merge_execution_intercept_callables, merge_guardrail_entries, merge_intercept_entries,
@@ -305,6 +305,35 @@ fn context_state_supports_extensions_events_and_builders() {
     assert_eq!(events.lock().unwrap().as_slice(), ["mark"]);
 }
 
+#[test]
+fn scope_end_metadata_merges_with_handle_metadata() {
+    let state = NemoRelayContextState::new();
+    let scope = state.create_scope_handle(
+        crate::api::scope::CreateScopeHandleParams::builder()
+            .name("agent")
+            .scope_type(ScopeType::Agent)
+            .metadata(json!({"a": 1, "b": 2, "c": 3}))
+            .build(),
+    );
+
+    let scope_end = state.build_scope_end_event(
+        EndScopeHandleParams::builder()
+            .handle(&scope)
+            .metadata(json!({"c": 3.5, "d": 4}))
+            .build(),
+    );
+    assert_eq!(
+        scope_end.metadata(),
+        Some(&json!({"a": 1, "b": 2, "c": 3.5, "d": 4}))
+    );
+
+    let scope_end = state.end_scope_handle(&scope, None, Some(json!({"c": 5, "e": 6})));
+    assert_eq!(
+        scope_end.metadata(),
+        Some(&json!({"a": 1, "b": 2, "c": 5, "e": 6}))
+    );
+}
+
 #[test]
 fn global_context_is_a_singleton_handle() {
     let first = global_context();