fix: map FlowError to OpenTelemetry error types

crates/core/src/api/llm.rs

@@ -34,9 +34,9 @@ use crate::api::runtime::{ScopeStackHandle, current_scope_stack};
 use crate::api::scope::event;
 use crate::api::scope::{EmitMarkEventParams, ScopeHandle};
 use crate::api::shared::{
-    ensure_runtime_owner, inject_dynamo_session_ids, metadata_with_otel_status,
-    resolve_parent_uuid, run_request_intercepts_with_codec_and_recorder, snapshot_event_sanitizers,
-    snapshot_event_subscribers,
+    ensure_runtime_owner, inject_dynamo_session_ids, metadata_with_otel_error,
+    metadata_with_otel_status, resolve_parent_uuid, run_request_intercepts_with_codec_and_recorder,
+    snapshot_event_sanitizers, snapshot_event_subscribers,
 };
 use crate::codec::request::{AnnotatedLlmRequest, Message};
 use crate::codec::response::{AnnotatedLlmResponse, attach_estimated_cost_for_provider};
@@ -1522,8 +1522,7 @@ pub async fn llm_call_execute(params: LlmCallExecuteParams) -> Result<Json> {
             Ok(response)
         }
         Err(error) => {
-            let end_metadata =
-                metadata_with_otel_status(metadata, "ERROR", Some(error.to_string()));
+            let end_metadata = metadata_with_otel_error(metadata, &error);
             let _ = emit_llm_end_without_output(
                 &handle,
                 end_metadata,
@@ -1729,8 +1728,7 @@ pub async fn llm_stream_call_execute(params: LlmStreamCallExecuteParams) -> Resu
             Ok(LlmJsonStream::from_closeable(wrapper))
         }
         Err(error) => {
-            let end_metadata =
-                metadata_with_otel_status(metadata, "ERROR", Some(error.to_string()));
+            let end_metadata = metadata_with_otel_error(metadata, &error);
             let _ = emit_llm_end_without_output(
                 &handle,
                 end_metadata,

crates/core/src/api/shared.rs

@@ -214,6 +214,16 @@ pub(crate) fn metadata_with_otel_status(
     metadata
 }
 
+pub(crate) fn metadata_with_otel_error(metadata: Option<Json>, error: &FlowError) -> Option<Json> {
+    let mut metadata = metadata_with_otel_status(metadata, "ERROR", Some(error.to_string()));
+    if let Some(Json::Object(metadata)) = metadata.as_mut() {
+        metadata
+            .entry("error.type".to_string())
+            .or_insert_with(|| Json::String(error.otel_error_type().to_string()));
+    }
+    metadata
+}
+
 pub(crate) type InterceptedLlmRequest = (
     LlmRequest,
     Option<Arc<AnnotatedLlmRequest>>,

crates/core/src/api/tool.rs

@@ -17,7 +17,7 @@ use crate::api::runtime::{
 use crate::api::scope::event;
 use crate::api::scope::{EmitMarkEventParams, ScopeHandle};
 use crate::api::shared::{
-    ensure_runtime_owner, metadata_with_otel_status, resolve_parent_uuid,
+    ensure_runtime_owner, metadata_with_otel_error, metadata_with_otel_status, resolve_parent_uuid,
     snapshot_event_sanitizers, snapshot_event_subscribers,
 };
 use crate::api::skill_load;
@@ -812,8 +812,7 @@ pub async fn tool_call_execute(params: ToolCallExecuteParams) -> Result<Json> {
             Ok(result)
         }
         Err(error) => {
-            let end_metadata =
-                metadata_with_otel_status(metadata, "ERROR", Some(error.to_string()));
+            let end_metadata = metadata_with_otel_error(metadata, &error);
             let _ = emit_tool_end_without_output(
                 &handle,
                 end_metadata,

crates/core/src/error.rs

@@ -134,6 +134,35 @@ pub enum FlowError {
 /// A specialized [`Result`](std::result::Result) type for NeMo Relay operations.
 pub type Result<T> = std::result::Result<T, FlowError>;
 
+impl FlowError {
+    /// Returns a low-cardinality classification suitable for OpenTelemetry's
+    /// `error.type` attribute.
+    ///
+    /// Relay-owned failures use stable `snake_case` codes. Internal failures
+    /// collapse to `internal_error` because Relay cannot reliably infer an
+    /// application exception type from an error message.
+    pub(crate) fn otel_error_type(&self) -> &str {
+        match self {
+            Self::AlreadyExists(_) => "already_exists",
+            Self::NotFound(_) => "not_found",
+            Self::InvalidArgument(_) => "invalid_argument",
+            Self::ScopeStackEmpty => "scope_stack_empty",
+            Self::GuardrailRejected(_) => "guardrail_rejected",
+            Self::Upstream(failure) => match failure.class {
+                UpstreamFailureClass::Connection => "connection_error",
+                UpstreamFailureClass::Timeout => "timeout",
+                UpstreamFailureClass::RetryableStatus => "retryable_status",
+                UpstreamFailureClass::ContextWindow => "context_window",
+                UpstreamFailureClass::ModelUnavailable => "model_unavailable",
+                UpstreamFailureClass::Authentication => "authentication",
+                UpstreamFailureClass::InvalidRequest => "invalid_request",
+                UpstreamFailureClass::Other => "upstream_error",
+            },
+            Self::Internal(_) => "internal_error",
+        }
+    }
+}
+
 #[cfg(test)]
 #[path = "../tests/coverage/error_tests.rs"]
 mod tests;

crates/core/src/stream.rs

@@ -44,7 +44,9 @@ use crate::api::runtime::{
     EventSubscriberFn, LlmJsonStream, LlmStreamInner, ScopeStackHandle, TASK_SCOPE_STACK,
     current_scope_stack,
 };
-use crate::api::shared::{metadata_with_otel_status, snapshot_event_sanitizers};
+use crate::api::shared::{
+    metadata_with_otel_error, metadata_with_otel_status, snapshot_event_sanitizers,
+};
 use crate::codec::response::{AnnotatedLlmResponse, attach_estimated_cost_for_provider};
 use crate::codec::traits::LlmResponseCodec;
 use crate::error::{FlowError, Result};
@@ -212,6 +214,16 @@ impl LlmStreamWrapper {
         self.finalization = self.emit_end_event(metadata, interrupted, false);
     }
 
+    fn finish_with_error(&mut self, error: &FlowError, interrupted: bool) {
+        if self.ended {
+            return;
+        }
+        self.ended = true;
+        self.inner.terminalize();
+        let metadata = metadata_with_otel_error(self.metadata.clone(), error);
+        self.finalization = self.emit_end_event(metadata, interrupted, false);
+    }
+
     /// Emit the LLM END event with aggregated response data.
     ///
     /// Calls the finalizer to produce the aggregated response, runs sanitize
@@ -446,17 +458,15 @@ impl Stream for LlmStreamWrapper {
                 match (this.collector)(raw_chunk.clone()) {
                     Ok(()) => Poll::Ready(Some(Ok(raw_chunk))),
                     Err(e) => {
-                        let message = e.to_string();
+                        this.finish_with_error(&e, true);
                         this.terminal_result = Some(Err(e));
-                        this.finish_with_status("ERROR", Some(message), true);
                         self.poll_next(cx)
                     }
                 }
             }
             Poll::Ready(Some(Err(e))) => {
-                let message = e.to_string();
+                this.finish_with_error(&e, true);
                 this.terminal_result = Some(Err(e));
-                this.finish_with_status("ERROR", Some(message), true);
                 self.poll_next(cx)
             }
             Poll::Ready(None) => {

crates/core/tests/coverage/error_tests.rs

@@ -54,6 +54,58 @@ fn test_error_debug() {
     assert!(debug.contains("AlreadyExists"));
 }
 
+#[test]
+fn otel_error_type_maps_relay_variants() {
+    assert_eq!(
+        FlowError::AlreadyExists("duplicate".into()).otel_error_type(),
+        "already_exists"
+    );
+    assert_eq!(
+        FlowError::NotFound("missing".into()).otel_error_type(),
+        "not_found"
+    );
+    assert_eq!(
+        FlowError::InvalidArgument("bad scope".into()).otel_error_type(),
+        "invalid_argument"
+    );
+    assert_eq!(
+        FlowError::ScopeStackEmpty.otel_error_type(),
+        "scope_stack_empty"
+    );
+    assert_eq!(
+        FlowError::GuardrailRejected("blocked".into()).otel_error_type(),
+        "guardrail_rejected"
+    );
+
+    let upstream_cases = [
+        (UpstreamFailureClass::Connection, "connection_error"),
+        (UpstreamFailureClass::Timeout, "timeout"),
+        (UpstreamFailureClass::RetryableStatus, "retryable_status"),
+        (UpstreamFailureClass::ContextWindow, "context_window"),
+        (UpstreamFailureClass::ModelUnavailable, "model_unavailable"),
+        (UpstreamFailureClass::Authentication, "authentication"),
+        (UpstreamFailureClass::InvalidRequest, "invalid_request"),
+        (UpstreamFailureClass::Other, "upstream_error"),
+    ];
+    for (class, expected) in upstream_cases {
+        let failure = UpstreamFailure {
+            status: None,
+            body: "provider failed".into(),
+            headers: std::collections::BTreeMap::new(),
+            class,
+        };
+        assert_eq!(FlowError::Upstream(failure).otel_error_type(), expected);
+    }
+}
+
+#[test]
+fn otel_error_type_maps_internal_failures_to_generic_code() {
+    assert_eq!(
+        FlowError::Internal("application callback failed".into()).otel_error_type(),
+        "internal_error"
+    );
+}
+
 #[test]
 fn upstream_failures_classify_retryability_and_render_status() {
     use std::collections::BTreeMap;

crates/core/tests/unit/llm_api_tests.rs

@@ -1718,6 +1718,7 @@ fn llm_call_execute_adds_otel_status_metadata_to_end_events() {
     let error_metadata = metadata_for("llm-error");
     assert_eq!(error_metadata["caller"], json!("llm-error"));
     assert_eq!(error_metadata["otel.status_code"], json!("ERROR"));
+    assert_eq!(error_metadata["error.type"], json!("internal_error"));
     assert!(
         error_metadata["otel.status_description"]
             .as_str()
@@ -1875,6 +1876,10 @@ fn llm_stream_call_execute_adds_otel_error_metadata_to_failed_end_events() {
         json!("llm-stream-upstream-error")
     );
     assert_eq!(upstream_error_metadata["otel.status_code"], json!("ERROR"));
+    assert_eq!(
+        upstream_error_metadata["error.type"],
+        json!("internal_error")
+    );
     assert!(
         upstream_error_metadata["otel.status_description"]
             .as_str()
@@ -1888,6 +1893,10 @@ fn llm_stream_call_execute_adds_otel_error_metadata_to_failed_end_events() {
         json!("llm-stream-collector-error")
     );
     assert_eq!(collector_error_metadata["otel.status_code"], json!("ERROR"));
+    assert_eq!(
+        collector_error_metadata["error.type"],
+        json!("internal_error")
+    );
     assert!(
         collector_error_metadata["otel.status_description"]
             .as_str()

crates/core/tests/unit/observability/otel_tests.rs

@@ -1458,6 +1458,31 @@ fn gen_ai_projection_emits_normalized_response_attributes() {
     );
 }
 
+#[test]
+fn gen_ai_end_projection_preserves_explicit_error_type() {
+    let event = Event::Scope(ScopeEvent::new(
+        BaseEvent::builder()
+            .uuid(Uuid::now_v7())
+            .name("chat")
+            .metadata(json!({
+                "otel.status_code": "ERROR",
+                "otel.status_description": "invalid argument: invalid value",
+                "error.type": "invalid_argument",
+            }))
+            .build(),
+        ScopeCategory::End,
+        Vec::new(),
+        EventCategory::from(ScopeType::Llm),
+        None,
+    ));
+
+    let attributes = attr_map(&crate::observability::otel_genai::end_attributes(&event));
+    assert_eq!(
+        attributes.get("error.type"),
+        Some(&"invalid_argument".to_string())
+    );
+}
+
 #[test]
 fn gen_ai_projection_prefers_standard_names_and_normalized_provider_details() {
     let agent = make_start_event(

crates/core/tests/unit/shared_tests.rs

@@ -120,6 +120,30 @@ fn test_metadata_with_otel_status_only_describes_errors() {
     );
 }
 
+#[test]
+fn test_metadata_with_otel_error_adds_structured_error_type() {
+    let metadata = metadata_with_otel_error(
+        Some(json!({"caller": "shared-error"})),
+        &FlowError::Internal("provider timed out".into()),
+    )
+    .unwrap();
+
+    assert_eq!(metadata["otel.status_code"], json!("ERROR"));
+    assert_eq!(metadata["error.type"], json!("internal_error"));
+    assert_eq!(
+        metadata["otel.status_description"],
+        json!("internal error: provider timed out")
+    );
+
+    let explicit_metadata = metadata_with_otel_error(
+        Some(json!({"error.type": "provider_timeout"})),
+        &FlowError::Internal("provider timed out".into()),
+    )
+    .unwrap();
+
+    assert_eq!(explicit_metadata["error.type"], json!("provider_timeout"));
+}
+
 #[test]
 fn test_resolve_parent_uuid_snapshot_and_runtime_owner_helpers() {
     let _guard = lock_runtime_owner();

crates/core/tests/unit/tool_api_tests.rs

@@ -99,6 +99,7 @@ fn tool_call_execute_adds_otel_status_metadata_to_end_events() {
     let error_metadata = metadata_for("tool-error");
     assert_eq!(error_metadata["caller"], json!("tool-error"));
     assert_eq!(error_metadata["otel.status_code"], json!("ERROR"));
+    assert_eq!(error_metadata["error.type"], json!("internal_error"));
     assert!(
         error_metadata["otel.status_description"]
             .as_str()

crates/node/tests/llm_tests.mjs

@@ -321,6 +321,7 @@ describe('LLM execute', () => {
       assert.equal(errorEnd.metadata.caller, 'node-llm-error');
       assert.equal(errorEnd.metadata['otel.status_code'], 'ERROR');
       assert.match(errorEnd.metadata['otel.status_description'], /llm status failure/);
+      assert.equal(errorEnd.metadata['error.type'], 'internal_error');
     } finally {
       deregisterSubscriber('node_llm_status_metadata_sub');
     }

crates/node/tests/tools_tests.mjs

@@ -415,6 +415,7 @@ describe('Tool execute', () => {
       assert.equal(errorEnd.metadata.caller, 'node-tool-error');
       assert.equal(errorEnd.metadata['otel.status_code'], 'ERROR');
       assert.match(errorEnd.metadata['otel.status_description'], /tool status failure/);
+      assert.equal(errorEnd.metadata['error.type'], 'internal_error');
     } finally {
       deregisterSubscriber('node_tool_status_metadata_sub');
     }

docs/configure-plugins/observability/opentelemetry.mdx

@@ -128,6 +128,36 @@ descendants remain attached to their nearest exported ancestor. This
 projection never emits `nemo_relay.*` fields and does not export message
 content, tool payloads, or retrieval content.
 
+### Error Type Mapping
+
+For managed LLM, tool, and stream failures, NeMo Relay maps structured
+`FlowError` values to the OpenTelemetry `error.type` attribute:
+
+| Relay error | `error.type` |
+|---|---|
+| `AlreadyExists` | `already_exists` |
+| `NotFound` | `not_found` |
+| `InvalidArgument` | `invalid_argument` |
+| `ScopeStackEmpty` | `scope_stack_empty` |
+| `GuardrailRejected` | `guardrail_rejected` |
+| Upstream connection failure | `connection_error` |
+| Upstream timeout | `timeout` |
+| Upstream retryable status | `retryable_status` |
+| Upstream context-window failure | `context_window` |
+| Upstream model unavailable | `model_unavailable` |
+| Upstream authentication failure | `authentication` |
+| Upstream invalid request | `invalid_request` |
+| Other upstream failure | `upstream_error` |
+| `Internal` | `internal_error` |
+
+External application and callback exceptions that do not have a more specific
+`FlowError` classification are represented as `Internal` and emit
+`internal_error`. NeMo Relay does not inspect error messages to recover Python,
+JavaScript, or application-defined exception class names. When no structured
+`FlowError` is available, such as a cancellation or dropped execution, the
+GenAI projection emits `_OTHER`. Caller-provided `error.type` metadata takes
+precedence over the derived mapping.
+
 ## Direct Subscriber
 
 <Tabs>