diff --git a/crates/core/src/api/llm.rs b/crates/core/src/api/llm.rs index 2c2877944..f398ebe93 100644 --- a/crates/core/src/api/llm.rs +++ b/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 { 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, diff --git a/crates/core/src/api/shared.rs b/crates/core/src/api/shared.rs index 5d0652c8d..2f823a27d 100644 --- a/crates/core/src/api/shared.rs +++ b/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, error: &FlowError) -> Option { + 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>, diff --git a/crates/core/src/api/tool.rs b/crates/core/src/api/tool.rs index b2e55b141..61bd6f922 100644 --- a/crates/core/src/api/tool.rs +++ b/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 { 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, diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index 957290b0d..270f9737a 100644 --- a/crates/core/src/error.rs +++ b/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 = std::result::Result; +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; diff --git a/crates/core/src/stream.rs b/crates/core/src/stream.rs index f6d3ca7b9..dd99fde19 100644 --- a/crates/core/src/stream.rs +++ b/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) => { diff --git a/crates/core/tests/coverage/error_tests.rs b/crates/core/tests/coverage/error_tests.rs index d7e000fb0..55e823a72 100644 --- a/crates/core/tests/coverage/error_tests.rs +++ b/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; diff --git a/crates/core/tests/unit/llm_api_tests.rs b/crates/core/tests/unit/llm_api_tests.rs index 30a357465..62615e9e2 100644 --- a/crates/core/tests/unit/llm_api_tests.rs +++ b/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() diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index ab3bc17ec..b703ecf8e 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/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( diff --git a/crates/core/tests/unit/shared_tests.rs b/crates/core/tests/unit/shared_tests.rs index 5375f4334..2249ca9f7 100644 --- a/crates/core/tests/unit/shared_tests.rs +++ b/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(); diff --git a/crates/core/tests/unit/tool_api_tests.rs b/crates/core/tests/unit/tool_api_tests.rs index 7628c5d15..99a1c1f53 100644 --- a/crates/core/tests/unit/tool_api_tests.rs +++ b/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() diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index fbba803f7..afeabd0fd 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/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'); } diff --git a/crates/node/tests/tools_tests.mjs b/crates/node/tests/tools_tests.mjs index c7ce4443d..b4d5a1a79 100644 --- a/crates/node/tests/tools_tests.mjs +++ b/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'); } diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index 79ebefd23..baa01c1e2 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/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