From 0903b01a62bb739adbadaab325a023a7b8fbd7ef Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 10:21:31 -0400 Subject: [PATCH 1/6] fix: preserve callback error details in traces Signed-off-by: Will Killian --- crates/core/src/api/shared.rs | 5 + crates/core/src/error.rs | 19 ++- crates/core/src/observability/otel.rs | 57 ++++++++- crates/core/src/plugin/dynamic/native.rs | 4 +- crates/core/tests/coverage/error_tests.rs | 6 + .../tests/unit/observability/otel_tests.rs | 120 ++++++++++++++++++ crates/core/tests/unit/shared_tests.rs | 11 ++ crates/ffi/src/error.rs | 4 +- crates/node/src/callback_factory.rs | 12 +- crates/node/src/promise_call.rs | 6 +- crates/node/tests/llm_tests.mjs | 1 + crates/node/tests/tools_tests.mjs | 1 + crates/python/src/py_callable.rs | 24 +++- .../observability/opentelemetry.mdx | 16 ++- python/tests/test_tools.py | 4 +- 15 files changed, 268 insertions(+), 22 deletions(-) diff --git a/crates/core/src/api/shared.rs b/crates/core/src/api/shared.rs index 2f823a27d..dadd33c69 100644 --- a/crates/core/src/api/shared.rs +++ b/crates/core/src/api/shared.rs @@ -220,6 +220,11 @@ pub(crate) fn metadata_with_otel_error(metadata: Option, error: &FlowError metadata .entry("error.type".to_string()) .or_insert_with(|| Json::String(error.otel_error_type().to_string())); + if let Some(exception_type) = error.exception_type() { + metadata + .entry("exception.type".to_string()) + .or_insert_with(|| Json::String(exception_type.to_string())); + } } metadata } diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index 270f9737a..23ed1c01a 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -129,6 +129,15 @@ pub enum FlowError { /// An internal runtime error (e.g., lock poisoning). #[error("internal error: {0}")] Internal(String), + + /// An exception raised by a language-binding callback. + #[error("internal error: {message}")] + CallbackException { + /// Original binding-rendered exception message. + message: String, + /// Original language exception class name. + exception_type: String, + }, } /// A specialized [`Result`](std::result::Result) type for NeMo Relay operations. @@ -158,7 +167,15 @@ impl FlowError { UpstreamFailureClass::InvalidRequest => "invalid_request", UpstreamFailureClass::Other => "upstream_error", }, - Self::Internal(_) => "internal_error", + Self::Internal(_) | Self::CallbackException { .. } => "internal_error", + } + } + + /// Returns the originating language exception class, when available. + pub(crate) fn exception_type(&self) -> Option<&str> { + match self { + Self::CallbackException { exception_type, .. } => Some(exception_type), + _ => None, } } } diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 362ea5ca3..c3ca0b1ba 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -948,6 +948,8 @@ pub(super) struct ActiveSpan { span_context: SpanContext, start_model_name: Option, projected_attributes: Vec, + descendant_error_type: Option, + descendant_exception_type: Option, } pub(super) struct OtelEventProcessor { @@ -1151,6 +1153,8 @@ impl OtelEventProcessor { span_context, start_model_name, projected_attributes, + descendant_error_type: None, + descendant_exception_type: None, }, ); } @@ -1167,6 +1171,36 @@ impl OtelEventProcessor { OpenTelemetryType::GenAi => super::otel_genai::end_attributes(event), OpenTelemetryType::OpenInference => super::openinference::end_attributes(event), }; + let is_error = metadata_string(event, "otel.status_code") == Some("ERROR"); + let explicit_error_type = metadata_string(event, "error.type"); + let error_type = is_error.then(|| { + explicit_error_type + .map(ToOwned::to_owned) + .or(active_span.descendant_error_type.take()) + .unwrap_or_else(|| "_OTHER".to_string()) + }); + let exception_type = is_error + .then(|| { + metadata_string(event, "exception.type") + .map(ToOwned::to_owned) + .or(active_span.descendant_exception_type.take()) + }) + .flatten(); + if matches!( + self.otel_type, + OpenTelemetryType::Full | OpenTelemetryType::GenAi + ) && let Some(error_type) = error_type.as_ref() + { + attributes.retain(|attribute| attribute.key.as_str() != "error.type"); + attributes.push(KeyValue::new("error.type", error_type.clone())); + } + if let Some(exception_type) = exception_type.as_ref() { + active_span.span.add_event_with_timestamp( + "exception", + to_system_time(*event.timestamp()), + vec![KeyValue::new("exception.type", exception_type.clone())], + ); + } let end_model_name = model_name_for_llm_event(event).or_else(|| active_span.start_model_name.take()); if self.otel_type == OpenTelemetryType::Full @@ -1187,6 +1221,14 @@ impl OtelEventProcessor { &self.attribute_mappings, )); } + if is_error && let Some(parent_span) = self.find_parent_span_mut(event) { + if parent_span.descendant_error_type.is_none() { + parent_span.descendant_error_type = error_type; + } + if parent_span.descendant_exception_type.is_none() { + parent_span.descendant_exception_type = exception_type; + } + } active_span.span.set_attributes(attributes); active_span .span @@ -1310,9 +1352,14 @@ impl OtelEventProcessor { } fn parent_span_uuid(&self, event: &Event) -> Option { - event - .parent_uuid() - .filter(|uuid| self.active_spans.contains_key(uuid)) + let parent_uuid = event.parent_uuid()?; + if self.active_spans.contains_key(&parent_uuid) { + return Some(parent_uuid); + } + let suppressed_parent = self.suppressed_parent_contexts.get(&parent_uuid)?; + self.active_spans.iter().find_map(|(uuid, active_span)| { + (active_span.span_context.span_id() == suppressed_parent.span_id()).then_some(*uuid) + }) } fn find_parent_span(&self, event: &Event) -> Option<&ActiveSpan> { @@ -1368,6 +1415,10 @@ impl OtelEventProcessor { } } +fn metadata_string<'a>(event: &'a Event, key: &str) -> Option<&'a str> { + event.metadata()?.get(key)?.as_str() +} + fn span_kind(event: &Event) -> SpanKind { match semantic_scope_type(event) { Some(ScopeType::Llm) => SpanKind::Client, diff --git a/crates/core/src/plugin/dynamic/native.rs b/crates/core/src/plugin/dynamic/native.rs index 3b0124b5b..7fe3d1998 100644 --- a/crates/core/src/plugin/dynamic/native.rs +++ b/crates/core/src/plugin/dynamic/native.rs @@ -1293,7 +1293,9 @@ fn status_from_flow_error(err: FlowError) -> NemoRelayStatus { FlowError::InvalidArgument(_) => NemoRelayStatus::InvalidArg, FlowError::ScopeStackEmpty => NemoRelayStatus::ScopeStackEmpty, FlowError::GuardrailRejected(_) => NemoRelayStatus::GuardrailRejected, - FlowError::Upstream(_) | FlowError::Internal(_) => NemoRelayStatus::Internal, + FlowError::Upstream(_) | FlowError::Internal(_) | FlowError::CallbackException { .. } => { + NemoRelayStatus::Internal + } } } diff --git a/crates/core/tests/coverage/error_tests.rs b/crates/core/tests/coverage/error_tests.rs index 55e823a72..5c4f02356 100644 --- a/crates/core/tests/coverage/error_tests.rs +++ b/crates/core/tests/coverage/error_tests.rs @@ -104,6 +104,12 @@ fn otel_error_type_maps_internal_failures_to_generic_code() { FlowError::Internal("application callback failed".into()).otel_error_type(), "internal_error" ); + let external = FlowError::CallbackException { + message: "ValueError: boom".into(), + exception_type: "ValueError".into(), + }; + assert_eq!(external.otel_error_type(), "internal_error"); + assert_eq!(external.exception_type(), Some("ValueError")); } #[test] diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 88bf5ecea..7ce23d747 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -528,6 +528,27 @@ fn make_end_event( ) } +fn make_end_event_with_metadata( + uuid: Uuid, + parent_uuid: Option, + name: &str, + scope_type: ScopeType, + metadata: Json, +) -> Event { + Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .parent_uuid_opt(parent_uuid) + .uuid(uuid) + .name(name) + .metadata(metadata) + .build(), + ScopeCategory::End, + Vec::new(), + EventCategory::from(scope_type), + None, + )) +} + fn make_scope_event( scope_category: ScopeCategory, uuid: Uuid, @@ -1581,6 +1602,105 @@ fn gen_ai_end_projection_preserves_explicit_error_type() { ); } +#[test] +fn failed_descendant_classification_and_exception_propagate_to_agent_span() { + for otel_type in [OpenTelemetryType::Full, OpenTelemetryType::GenAi] { + let (provider, exporter) = make_provider(); + let mut processor = + OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings( + provider, + "error-propagation-test".to_string(), + otel_type, + MarkProjection::default(), + default_mark_exclude_names(), + Vec::new(), + ); + let agent_uuid = Uuid::now_v7(); + let function_uuid = Uuid::now_v7(); + let llm_uuid = Uuid::now_v7(); + processor.process(&make_start_event( + agent_uuid, + None, + "agent", + ScopeType::Agent, + None, + )); + let llm_parent_uuid = if otel_type == OpenTelemetryType::GenAi { + processor.process(&make_start_event( + function_uuid, + Some(agent_uuid), + "function", + ScopeType::Function, + None, + )); + function_uuid + } else { + agent_uuid + }; + processor.process(&make_start_event( + llm_uuid, + Some(llm_parent_uuid), + "chat", + ScopeType::Llm, + None, + )); + processor.process(&make_end_event_with_metadata( + llm_uuid, + Some(llm_parent_uuid), + "chat", + ScopeType::Llm, + json!({ + "otel.status_code": "ERROR", + "otel.status_description": "internal error: ValueError: boom", + "error.type": "internal_error", + "exception.type": "ValueError", + }), + )); + if otel_type == OpenTelemetryType::GenAi { + processor.process(&make_end_event_with_metadata( + function_uuid, + Some(agent_uuid), + "function", + ScopeType::Function, + json!({ + "otel.status_code": "ERROR", + "otel.status_description": "internal error: ValueError: boom", + }), + )); + } + processor.process(&make_end_event_with_metadata( + agent_uuid, + None, + "agent", + ScopeType::Agent, + json!({ + "otel.status_code": "ERROR", + "otel.status_description": "internal error: ValueError: boom", + }), + )); + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 2); + for span in &spans { + assert_eq!( + attr_map(&span.attributes).get("error.type"), + Some(&"internal_error".to_string()) + ); + let exception = span + .events + .events + .iter() + .find(|event| event.name.as_ref() == "exception") + .expect("expected exception event"); + assert_eq!( + attr_map(&exception.attributes).get("exception.type"), + Some(&"ValueError".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 2249ca9f7..720b8d769 100644 --- a/crates/core/tests/unit/shared_tests.rs +++ b/crates/core/tests/unit/shared_tests.rs @@ -142,6 +142,17 @@ fn test_metadata_with_otel_error_adds_structured_error_type() { .unwrap(); assert_eq!(explicit_metadata["error.type"], json!("provider_timeout")); + + let external_metadata = metadata_with_otel_error( + None, + &FlowError::CallbackException { + message: "ValueError: boom".into(), + exception_type: "ValueError".into(), + }, + ) + .unwrap(); + assert_eq!(external_metadata["error.type"], json!("internal_error")); + assert_eq!(external_metadata["exception.type"], json!("ValueError")); } #[test] diff --git a/crates/ffi/src/error.rs b/crates/ffi/src/error.rs index 207197123..3bfa7a4f2 100644 --- a/crates/ffi/src/error.rs +++ b/crates/ffi/src/error.rs @@ -120,7 +120,9 @@ impl From<&FlowError> for NemoRelayStatus { FlowError::InvalidArgument(_) => NemoRelayStatus::InvalidArg, FlowError::ScopeStackEmpty => NemoRelayStatus::ScopeStackEmpty, FlowError::GuardrailRejected(_) => NemoRelayStatus::GuardrailRejected, - FlowError::Upstream(_) | FlowError::Internal(_) => NemoRelayStatus::Internal, + FlowError::Upstream(_) + | FlowError::Internal(_) + | FlowError::CallbackException { .. } => NemoRelayStatus::Internal, } } } diff --git a/crates/node/src/callback_factory.rs b/crates/node/src/callback_factory.rs index cb7988849..8f4464d71 100644 --- a/crates/node/src/callback_factory.rs +++ b/crates/node/src/callback_factory.rs @@ -165,6 +165,7 @@ const CALLBACK_FACTORIES_SOURCE: &str = r#"(() => { }, (error) => { settlePublication(); let message = 'unknown error'; + let exceptionType = 'Error'; try { if (typeof error === 'string') { message = error; @@ -173,8 +174,11 @@ const CALLBACK_FACTORIES_SOURCE: &str = r#"(() => { } else if (error != null && typeof error.message === 'string') { message = error.message; } + if (error != null && typeof error.name === 'string' && error.name.length > 0) { + exceptionType = error.name; + } } catch {} - reject(message); + reject(message, exceptionType); }); }; eventSanitizerContext.run(token, invoke); @@ -212,11 +216,15 @@ const CALLBACK_FACTORIES_SOURCE: &str = r#"(() => { ) { if (error != null) { let message = 'unknown error'; + let exceptionType = 'Error'; try { message = String(error?.message ?? error); + if (typeof error?.name === 'string' && error.name.length > 0) { + exceptionType = error.name; + } } catch {} if (typeof reject === 'function') { - reject(message); + reject(message, exceptionType); } return; } diff --git a/crates/node/src/promise_call.rs b/crates/node/src/promise_call.rs index 892bd6627..7832468d5 100644 --- a/crates/node/src/promise_call.rs +++ b/crates/node/src/promise_call.rs @@ -334,7 +334,11 @@ fn build_completion_unknowns( let message = ctx .get::(0) .unwrap_or_else(|_| "unknown error".to_string()); - completion.send(Err(FlowError::Internal(message))); + let exception_type = ctx.get::(1).unwrap_or_else(|_| "Error".to_string()); + completion.send(Err(FlowError::CallbackException { + message, + exception_type, + })); ctx.env.get_undefined() })?; diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index a2254af50..46c3a0f7b 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/crates/node/tests/llm_tests.mjs @@ -322,6 +322,7 @@ describe('LLM execute', () => { 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'); + assert.equal(errorEnd.metadata['exception.type'], '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 29a262e91..d4cdf5acf 100644 --- a/crates/node/tests/tools_tests.mjs +++ b/crates/node/tests/tools_tests.mjs @@ -416,6 +416,7 @@ describe('Tool execute', () => { 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'); + assert.equal(errorEnd.metadata['exception.type'], 'Error'); } finally { deregisterSubscriber('node_tool_status_metadata_sub'); } diff --git a/crates/python/src/py_callable.rs b/crates/python/src/py_callable.rs index 50772a3e6..323783058 100644 --- a/crates/python/src/py_callable.rs +++ b/crates/python/src/py_callable.rs @@ -59,6 +59,20 @@ use crate::py_types::{ type PyValueFuture = Pin>> + Send>>; +fn python_callback_error(error: PyErr) -> FlowError { + let exception_type = Python::attach(|py| { + error + .get_type(py) + .getattr("__name__") + .and_then(|name| name.extract::()) + .unwrap_or_else(|_| "Exception".to_string()) + }); + FlowError::CallbackException { + message: error.to_string(), + exception_type, + } +} + struct CancellablePyFuture { inner: PyValueFuture, scheduled: Arc>, @@ -296,9 +310,7 @@ async fn resolve_json_or_future( match outcome? { Ok(json) => Ok(json), Err(future) => { - let py_result = future - .await - .map_err(|e| FlowError::Internal(e.to_string()))?; + let py_result = future.await.map_err(python_callback_error)?; Python::attach(|py| { py_to_json(py_result.bind(py)) .map_err(|e: PyErr| FlowError::Internal(e.to_string())) @@ -573,7 +585,7 @@ async fn await_async_iter_task_result(task: Py) -> FlowResult context.call_method1("run", (callback.bind(py), py_args)), None => callback.bind(py).call1((py_args,)), } - .map_err(|error| FlowError::Internal(error.to_string()))?; + .map_err(python_callback_error)?; split_json_or_future_with_locals(py, result.unbind(), task_locals.as_ref()) })) .await @@ -1508,7 +1520,7 @@ pub fn wrap_py_llm_exec_fn( Some(context) => context.call_method1("run", (callback.bind(py), py_req)), None => callback.bind(py).call1((py_req,)), } - .map_err(|error| FlowError::Internal(error.to_string()))?; + .map_err(python_callback_error)?; split_json_or_future_with_locals(py, result.unbind(), task_locals.as_ref()) })) .await diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index e04e4c98b..47b2b4012 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -210,14 +210,18 @@ For managed LLM, tool, and stream failures, NeMo Relay maps structured | Upstream invalid request | `invalid_request` | | Other upstream failure | `upstream_error` | | `Internal` | `internal_error` | +| Binding callback exception | `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. +`FlowError` classification emit `internal_error`. Python and JavaScript callback +boundaries also preserve the exception class separately, and both the `full` +and `gen_ai` projections emit an `exception` span event with `exception.type`. +NeMo Relay does not inspect error messages to recover exception class names. +When an errored parent span has no useful classification of its own, it +inherits the failed descendant's `error.type` and exception type. When no +structured `FlowError` is available, such as a cancellation or dropped +execution, the projection emits `_OTHER`. Caller-provided `error.type` metadata +takes precedence over the derived mapping. ## Direct Subscriber diff --git a/python/tests/test_tools.py b/python/tests/test_tools.py index 51de3c645..96b72affb 100644 --- a/python/tests/test_tools.py +++ b/python/tests/test_tools.py @@ -208,7 +208,7 @@ async def test_execute_failure_emits_end_event(self): subscribers.register("py_tool_exec_failure_sub", lambda e: events.append(e)) def failing(args): - raise RuntimeError("boom") + raise ValueError("boom") with pytest.raises(RuntimeError, match="boom"): await tools.execute("failing_tool", {"x": 1}, failing) @@ -224,6 +224,8 @@ def failing(args): assert all(e.category == "tool" for e in events) assert events[0].uuid == events[1].uuid assert events[1].data is None + assert events[1].metadata["error.type"] == "internal_error" + assert events[1].metadata["exception.type"] == "ValueError" class TestToolGuardrails: From a5522cfa7cb6a194c76affbde8e88321580a4fe4 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 12:11:27 -0400 Subject: [PATCH 2/6] fix: address observability review feedback Signed-off-by: Will Killian --- crates/core/src/observability/otel.rs | 4 +- .../tests/unit/observability/otel_tests.rs | 124 ++++++++++++++++++ crates/core/tests/unit/shared_tests.rs | 13 ++ crates/node/src/callback_factory.rs | 21 ++- crates/node/tests/llm_tests.mjs | 11 +- crates/node/tests/tools_tests.mjs | 4 +- .../observability/opentelemetry.mdx | 5 + 7 files changed, 169 insertions(+), 13 deletions(-) diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index c3ca0b1ba..e32895627 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -1358,7 +1358,9 @@ impl OtelEventProcessor { } let suppressed_parent = self.suppressed_parent_contexts.get(&parent_uuid)?; self.active_spans.iter().find_map(|(uuid, active_span)| { - (active_span.span_context.span_id() == suppressed_parent.span_id()).then_some(*uuid) + (active_span.span_context.trace_id() == suppressed_parent.trace_id() + && active_span.span_context.span_id() == suppressed_parent.span_id()) + .then_some(*uuid) }) } diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 7ce23d747..12ef6aa4a 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -1701,6 +1701,130 @@ fn failed_descendant_classification_and_exception_propagate_to_agent_span() { } } +#[test] +fn suppressed_parent_error_propagation_isolated_by_trace_id() { + let (provider, exporter) = make_provider(); + let mut processor = OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings( + provider, + "error-trace-isolation-test".to_string(), + OpenTelemetryType::GenAi, + MarkProjection::default(), + default_mark_exclude_names(), + Vec::new(), + ); + let shared_span_id = [0xAB; 8]; + let mut first_agent_bytes = [0x11; 16]; + first_agent_bytes[8..].copy_from_slice(&shared_span_id); + let mut second_agent_bytes = [0x22; 16]; + second_agent_bytes[8..].copy_from_slice(&shared_span_id); + let cases = [ + ( + Uuid::from_bytes(first_agent_bytes), + Uuid::now_v7(), + Uuid::now_v7(), + "first_error", + "FirstException", + ), + ( + Uuid::from_bytes(second_agent_bytes), + Uuid::now_v7(), + Uuid::now_v7(), + "second_error", + "SecondException", + ), + ]; + assert_eq!( + relay_span_id(cases[0].0), + relay_span_id(cases[1].0), + "fixture agents must share a span ID" + ); + assert_ne!( + relay_trace_id(cases[0].0), + relay_trace_id(cases[1].0), + "fixture agents must belong to different traces" + ); + + for (agent_uuid, function_uuid, llm_uuid, _, _) in cases { + processor.process(&make_start_event( + agent_uuid, + None, + "agent", + ScopeType::Agent, + None, + )); + processor.process(&make_start_event( + function_uuid, + Some(agent_uuid), + "function", + ScopeType::Function, + None, + )); + processor.process(&make_start_event( + llm_uuid, + Some(function_uuid), + "chat", + ScopeType::Llm, + None, + )); + } + for (agent_uuid, function_uuid, llm_uuid, error_type, exception_type) in cases { + processor.process(&make_end_event_with_metadata( + llm_uuid, + Some(function_uuid), + "chat", + ScopeType::Llm, + json!({ + "otel.status_code": "ERROR", + "error.type": error_type, + "exception.type": exception_type, + }), + )); + processor.process(&make_end_event_with_metadata( + function_uuid, + Some(agent_uuid), + "function", + ScopeType::Function, + json!({"otel.status_code": "ERROR"}), + )); + } + for (agent_uuid, _, _, _, _) in cases { + processor.process(&make_end_event_with_metadata( + agent_uuid, + None, + "agent", + ScopeType::Agent, + json!({"otel.status_code": "ERROR"}), + )); + } + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 4); + for (agent_uuid, _, _, error_type, exception_type) in cases { + let agent_span = spans + .iter() + .find(|span| { + span.span_context.trace_id() == relay_trace_id(agent_uuid) + && span.parent_span_id == SpanId::INVALID + }) + .expect("expected agent span for trace"); + assert_eq!( + attr_map(&agent_span.attributes).get("error.type"), + Some(&error_type.to_string()) + ); + let exception = agent_span + .events + .events + .iter() + .find(|event| event.name.as_ref() == "exception") + .expect("expected propagated exception event"); + assert_eq!( + attr_map(&exception.attributes).get("exception.type"), + Some(&exception_type.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 720b8d769..69793a38f 100644 --- a/crates/core/tests/unit/shared_tests.rs +++ b/crates/core/tests/unit/shared_tests.rs @@ -153,6 +153,19 @@ fn test_metadata_with_otel_error_adds_structured_error_type() { .unwrap(); assert_eq!(external_metadata["error.type"], json!("internal_error")); assert_eq!(external_metadata["exception.type"], json!("ValueError")); + + let explicit_exception_metadata = metadata_with_otel_error( + Some(json!({"exception.type": "CallerException"})), + &FlowError::CallbackException { + message: "ValueError: boom".into(), + exception_type: "ValueError".into(), + }, + ) + .unwrap(); + assert_eq!( + explicit_exception_metadata["exception.type"], + json!("CallerException") + ); } #[test] diff --git a/crates/node/src/callback_factory.rs b/crates/node/src/callback_factory.rs index 8f4464d71..a50da1e3e 100644 --- a/crates/node/src/callback_factory.rs +++ b/crates/node/src/callback_factory.rs @@ -171,11 +171,17 @@ const CALLBACK_FACTORIES_SOURCE: &str = r#"(() => { message = error; } else if (error === null || (typeof error !== 'object' && typeof error !== 'function')) { message = String(error); - } else if (error != null && typeof error.message === 'string') { - message = error.message; + } else if (error != null) { + const errorMessage = error.message; + if (typeof errorMessage === 'string') { + message = errorMessage; + } } - if (error != null && typeof error.name === 'string' && error.name.length > 0) { - exceptionType = error.name; + } catch {} + try { + const errorName = error?.name; + if (typeof errorName === 'string' && errorName.length > 0) { + exceptionType = errorName; } } catch {} reject(message, exceptionType); @@ -219,8 +225,11 @@ const CALLBACK_FACTORIES_SOURCE: &str = r#"(() => { let exceptionType = 'Error'; try { message = String(error?.message ?? error); - if (typeof error?.name === 'string' && error.name.length > 0) { - exceptionType = error.name; + } catch {} + try { + const errorName = error?.name; + if (typeof errorName === 'string' && errorName.length > 0) { + exceptionType = errorName; } } catch {} if (typeof reject === 'function') { diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index 46c3a0f7b..b67a7a152 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/crates/node/tests/llm_tests.mjs @@ -90,6 +90,9 @@ function sparseArray() { function unprintableError() { const error = new Error('sanitize request guardrail failed'); Object.defineProperties(error, { + name: { + value: 'GetterError', + }, message: { get() { throw new Error('message getter boom'); @@ -289,7 +292,7 @@ describe('LLM execute', () => { 'exec_status_error_llm', makeNative(), async () => { - throw new Error('llm status failure'); + throw unprintableError(); }, null, null, @@ -299,7 +302,7 @@ describe('LLM execute', () => { }, null, ), - /llm status failure/, + /unknown error/, ); await flushSubscribers(); @@ -320,9 +323,9 @@ describe('LLM execute', () => { assert.ok(errorEnd, 'expected failed llm end event'); 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.match(errorEnd.metadata['otel.status_description'], /unknown error/); assert.equal(errorEnd.metadata['error.type'], 'internal_error'); - assert.equal(errorEnd.metadata['exception.type'], 'Error'); + assert.equal(errorEnd.metadata['exception.type'], 'GetterError'); } finally { deregisterSubscriber('node_llm_status_metadata_sub'); } diff --git a/crates/node/tests/tools_tests.mjs b/crates/node/tests/tools_tests.mjs index d4cdf5acf..a83b981b4 100644 --- a/crates/node/tests/tools_tests.mjs +++ b/crates/node/tests/tools_tests.mjs @@ -384,7 +384,7 @@ describe('Tool execute', () => { 'exec_status_error_tool', {}, async () => { - throw new Error('tool status failure'); + throw new TypeError('tool status failure'); }, null, null, @@ -416,7 +416,7 @@ describe('Tool execute', () => { 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'); - assert.equal(errorEnd.metadata['exception.type'], 'Error'); + assert.equal(errorEnd.metadata['exception.type'], 'TypeError'); } finally { deregisterSubscriber('node_tool_status_metadata_sub'); } diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index 47b2b4012..b78b8aa2b 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -223,6 +223,11 @@ structured `FlowError` is available, such as a cancellation or dropped execution, the projection emits `_OTHER`. Caller-provided `error.type` metadata takes precedence over the derived mapping. +`FlowError` is an exhaustive Rust enum. Rust callers upgrading to this release +must handle the new `CallbackException` variant in exhaustive matches. It maps +to the same internal status as `Internal`, while retaining `exception_type` for +observability projection. + ## Direct Subscriber From 4fb5d5d0f0fa3d8d5a7d9fd3978cb09694530384 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 12:29:52 -0400 Subject: [PATCH 3/6] docs: clarify exception type precedence Signed-off-by: Will Killian --- docs/configure-plugins/observability/opentelemetry.mdx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index b78b8aa2b..edfc432d7 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -220,8 +220,8 @@ NeMo Relay does not inspect error messages to recover exception class names. When an errored parent span has no useful classification of its own, it inherits the failed descendant's `error.type` and exception type. When no structured `FlowError` is available, such as a cancellation or dropped -execution, the projection emits `_OTHER`. Caller-provided `error.type` metadata -takes precedence over the derived mapping. +execution, the projection emits `_OTHER`. Caller-provided `error.type` and +`exception.type` metadata take precedence over values derived from `FlowError`. `FlowError` is an exhaustive Rust enum. Rust callers upgrading to this release must handle the new `CallbackException` variant in exhaustive matches. It maps From 83597d13f2ed08589373fdc81cc75f022c697a10 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 15:12:19 -0400 Subject: [PATCH 4/6] fix: preserve Python streaming callback types Signed-off-by: Will Killian --- crates/python/src/py_callable.rs | 8 ++-- python/tests/test_llm.py | 67 +++++++++++++++++++++++++++----- 2 files changed, 62 insertions(+), 13 deletions(-) diff --git a/crates/python/src/py_callable.rs b/crates/python/src/py_callable.rs index 323783058..5b3d513a2 100644 --- a/crates/python/src/py_callable.rs +++ b/crates/python/src/py_callable.rs @@ -517,7 +517,7 @@ async fn resolve_py_object_or_future( ) -> FlowResult> { match outcome? { Ok(value) => Ok(value), - Err(future) => future.await.map_err(|e| FlowError::Internal(e.to_string())), + Err(future) => future.await.map_err(python_callback_error), } } @@ -1558,7 +1558,7 @@ pub fn wrap_py_llm_stream_exec_fn( Some(context) => context.call_method1("run", (callback.bind(py), py_req)), None => callback.bind(py).call1((py_req,)), } - .map_err(|error| FlowError::Internal(error.to_string()))?; + .map_err(python_callback_error)?; let outcome = split_py_object_or_future_with_locals( py, result.unbind(), @@ -1578,7 +1578,7 @@ pub fn wrap_py_llm_stream_exec_fn( /// The collector is invoked with each intercepted chunk (after stream response /// intercepts have been applied). It receives a single JSON-converted Python /// object argument. If the Python callable raises an exception, it is converted -/// to a `FlowError::Internal` and returned as `Err`, which terminates the +/// to a `FlowError::CallbackException` and returned as `Err`, which terminates the /// stream. If the callable returns normally (including `None`), the collector /// returns `Ok(())`. pub fn wrap_py_collector_fn( @@ -1590,7 +1590,7 @@ pub fn wrap_py_collector_fn( .map_err(|e| FlowError::Internal(format!("collector json_to_py failed: {e}")))?; py_fn .call1(py, (py_chunk,)) - .map_err(|e| FlowError::Internal(format!("Python collector error: {e}")))?; + .map_err(python_callback_error)?; Ok(()) }) }) diff --git a/python/tests/test_llm.py b/python/tests/test_llm.py index 0065ed786..3e3686594 100644 --- a/python/tests/test_llm.py +++ b/python/tests/test_llm.py @@ -1339,21 +1339,70 @@ async def test_stream_execution_intercept_propagates_direct___anext__error(self) intercepts.deregister_llm_stream_execution("py_llm_stream_direct_error") async def test_stream_execute_collector_failure_raises(self): + events = [] + subscribers.register("py_llm_stream_collector_failure_sub", events.append) + def stream_func(request): async def gen(): yield {"token": "hello"} return gen() - stream = await llm.stream_execute( - "stream_collector_fail_llm", - make_request(), - stream_func, - lambda chunk: raise_runtime_error("collector boom"), - lambda: {}, - ) - with pytest.raises(RuntimeError, match="collector boom"): - await anext(stream) + try: + stream = await llm.stream_execute( + "stream_collector_fail_llm", + make_request(), + stream_func, + lambda chunk: raise_runtime_error("collector boom"), + lambda: {}, + ) + with pytest.raises(RuntimeError, match="collector boom"): + await anext(stream) + await subscribers.flush_async() + finally: + subscribers.deregister("py_llm_stream_collector_failure_sub") + + metadata = _llm_event(events, "stream_collector_fail_llm", "end").metadata + assert isinstance(metadata, dict) + assert metadata["exception.type"] == "RuntimeError" + + async def test_stream_execute_callback_failure_emits_exception_type(self): + events = [] + subscribers.register("py_llm_stream_callback_failure_sub", events.append) + + def stream_func(request): + raise ValueError("stream callback boom") + + async def async_stream_func(request): + raise TypeError("async stream callback boom") + + try: + with pytest.raises(RuntimeError, match="stream callback boom"): + await llm.stream_execute( + "stream_callback_fail_llm", + make_request(), + stream_func, + lambda chunk: None, + lambda: {}, + ) + with pytest.raises(RuntimeError, match="async stream callback boom"): + await llm.stream_execute( + "async_stream_callback_fail_llm", + make_request(), + async_stream_func, + lambda chunk: None, + lambda: {}, + ) + await subscribers.flush_async() + finally: + subscribers.deregister("py_llm_stream_callback_failure_sub") + + metadata = _llm_event(events, "stream_callback_fail_llm", "end").metadata + assert isinstance(metadata, dict) + assert metadata["exception.type"] == "ValueError" + metadata = _llm_event(events, "async_stream_callback_fail_llm", "end").metadata + assert isinstance(metadata, dict) + assert metadata["exception.type"] == "TypeError" async def test_stream_execute_finalizer_failure_records_null_output(self): events = [] From 78d6aea19ae56683b88d704229eb3588eb259f9f Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 15:33:44 -0400 Subject: [PATCH 5/6] fix: preserve callback types across omitted spans Signed-off-by: Will Killian --- crates/core/src/observability/otel.rs | 21 ++++++ .../tests/unit/observability/otel_tests.rs | 66 +++++++++++++++++++ crates/ffi/tests/coverage/error_tests.rs | 7 ++ crates/node/src/callable.rs | 9 ++- crates/node/src/callback_factory.rs | 14 +++- crates/node/tests/llm_tests.mjs | 8 +-- crates/node/tests/tools_tests.mjs | 4 +- crates/python/src/py_callable.rs | 2 +- python/tests/test_llm.py | 29 ++++++++ 9 files changed, 150 insertions(+), 10 deletions(-) diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index e32895627..5a80992de 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -1161,6 +1161,7 @@ impl OtelEventProcessor { fn process_end(&mut self, event: &Event) { let Some(mut active_span) = self.active_spans.remove(&event.uuid()) else { + self.propagate_suppressed_error_metadata(event); return; }; self.record_completed_span_context(event.uuid(), active_span.span_context.clone()); @@ -1235,6 +1236,26 @@ impl OtelEventProcessor { .end_with_timestamp(to_system_time(*event.timestamp())); } + fn propagate_suppressed_error_metadata(&mut self, event: &Event) { + if self.otel_type != OpenTelemetryType::GenAi + || !self.suppressed_parent_contexts.contains_key(&event.uuid()) + || metadata_string(event, "otel.status_code") != Some("ERROR") + { + return; + } + let error_type = metadata_string(event, "error.type").map(ToOwned::to_owned); + let exception_type = metadata_string(event, "exception.type").map(ToOwned::to_owned); + let Some(parent_span) = self.find_parent_span_mut(event) else { + return; + }; + if parent_span.descendant_error_type.is_none() { + parent_span.descendant_error_type = error_type; + } + if parent_span.descendant_exception_type.is_none() { + parent_span.descendant_exception_type = exception_type; + } + } + fn process_mark(&mut self, event: &Event) { if self.otel_type == OpenTelemetryType::GenAi { return; diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 12ef6aa4a..1d8172947 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -1701,6 +1701,72 @@ fn failed_descendant_classification_and_exception_propagate_to_agent_span() { } } +#[test] +fn suppressed_function_error_propagates_to_agent_span() { + let (provider, exporter) = make_provider(); + let mut processor = OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings( + provider, + "suppressed-error-propagation-test".to_string(), + OpenTelemetryType::GenAi, + MarkProjection::default(), + default_mark_exclude_names(), + Vec::new(), + ); + let agent_uuid = Uuid::now_v7(); + let function_uuid = Uuid::now_v7(); + processor.process(&make_start_event( + agent_uuid, + None, + "agent", + ScopeType::Agent, + None, + )); + processor.process(&make_start_event( + function_uuid, + Some(agent_uuid), + "function", + ScopeType::Function, + None, + )); + processor.process(&make_end_event_with_metadata( + function_uuid, + Some(agent_uuid), + "function", + ScopeType::Function, + json!({ + "otel.status_code": "ERROR", + "error.type": "internal_error", + "exception.type": "ValueError", + }), + )); + processor.process(&make_end_event_with_metadata( + agent_uuid, + None, + "agent", + ScopeType::Agent, + json!({"otel.status_code": "ERROR"}), + )); + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 1); + let agent_span = &spans[0]; + assert_eq!( + attr_map(&agent_span.attributes).get("error.type"), + Some(&"internal_error".to_string()) + ); + let exception = agent_span + .events + .events + .iter() + .find(|event| event.name.as_ref() == "exception") + .expect("expected propagated exception event"); + assert_eq!( + attr_map(&exception.attributes).get("exception.type"), + Some(&"ValueError".to_string()) + ); +} + #[test] fn suppressed_parent_error_propagation_isolated_by_trace_id() { let (provider, exporter) = make_provider(); diff --git a/crates/ffi/tests/coverage/error_tests.rs b/crates/ffi/tests/coverage/error_tests.rs index fe2792666..cc5587d47 100644 --- a/crates/ffi/tests/coverage/error_tests.rs +++ b/crates/ffi/tests/coverage/error_tests.rs @@ -70,6 +70,13 @@ fn test_status_from_error_maps_variants_and_sets_message() { FlowError::Internal("boom".into()), NemoRelayStatus::Internal, ), + ( + FlowError::CallbackException { + message: "callback boom".into(), + exception_type: "ValueError".into(), + }, + NemoRelayStatus::Internal, + ), (FlowError::ScopeStackEmpty, NemoRelayStatus::ScopeStackEmpty), ]; diff --git a/crates/node/src/callable.rs b/crates/node/src/callable.rs index b6052059a..e807979be 100644 --- a/crates/node/src/callable.rs +++ b/crates/node/src/callable.rs @@ -163,6 +163,8 @@ struct MiddlewareCallbackResult { value: Json, #[serde(default)] error: String, + #[serde(default, rename = "exceptionType")] + exception_type: String, } /// Wrap a middleware callback so exceptions cross the N-API boundary as data. @@ -207,11 +209,16 @@ pub(crate) fn unwrap_middleware_result(value: Json, error_prefix: &str) -> Resul })?; if result.ok { Ok(result.value) - } else { + } else if result.exception_type.is_empty() { Err(FlowError::Internal(format!( "{error_prefix}: {}", result.error ))) + } else { + Err(FlowError::CallbackException { + message: format!("{error_prefix}: {}", result.error), + exception_type: result.exception_type, + }) } } diff --git a/crates/node/src/callback_factory.rs b/crates/node/src/callback_factory.rs index a50da1e3e..0990d343e 100644 --- a/crates/node/src/callback_factory.rs +++ b/crates/node/src/callback_factory.rs @@ -198,10 +198,20 @@ const CALLBACK_FACTORIES_SOURCE: &str = r#"(() => { return { ok: true, value: jsonValue(value === undefined ? null : value) }; } catch (error) { let message = 'JavaScript callback failed'; + let exceptionType = 'Error'; try { - message = String(error?.message ?? error); + const errorMessage = error?.message; + if (typeof errorMessage === 'string') { + message = errorMessage; + } + } catch {} + try { + const errorName = error?.name; + if (typeof errorName === 'string' && errorName.length > 0) { + exceptionType = errorName; + } } catch {} - return { ok: false, error: message }; + return { ok: false, error: message, exceptionType }; } }; }, diff --git a/crates/node/tests/llm_tests.mjs b/crates/node/tests/llm_tests.mjs index b67a7a152..a65e3f702 100644 --- a/crates/node/tests/llm_tests.mjs +++ b/crates/node/tests/llm_tests.mjs @@ -288,10 +288,10 @@ describe('LLM execute', () => { await assert.rejects( () => - llmCallExecuteAsync( + llmCallExecute( 'exec_status_error_llm', makeNative(), - async () => { + () => { throw unprintableError(); }, null, @@ -302,7 +302,7 @@ describe('LLM execute', () => { }, null, ), - /unknown error/, + /JavaScript callback failed/, ); await flushSubscribers(); @@ -323,7 +323,7 @@ describe('LLM execute', () => { assert.ok(errorEnd, 'expected failed llm end event'); assert.equal(errorEnd.metadata.caller, 'node-llm-error'); assert.equal(errorEnd.metadata['otel.status_code'], 'ERROR'); - assert.match(errorEnd.metadata['otel.status_description'], /unknown error/); + assert.match(errorEnd.metadata['otel.status_description'], /JavaScript callback failed/); assert.equal(errorEnd.metadata['error.type'], 'internal_error'); assert.equal(errorEnd.metadata['exception.type'], 'GetterError'); } finally { diff --git a/crates/node/tests/tools_tests.mjs b/crates/node/tests/tools_tests.mjs index a83b981b4..e1cc7c6a1 100644 --- a/crates/node/tests/tools_tests.mjs +++ b/crates/node/tests/tools_tests.mjs @@ -380,10 +380,10 @@ describe('Tool execute', () => { await assert.rejects( () => - toolCallExecuteAsync( + toolCallExecute( 'exec_status_error_tool', {}, - async () => { + () => { throw new TypeError('tool status failure'); }, null, diff --git a/crates/python/src/py_callable.rs b/crates/python/src/py_callable.rs index 5b3d513a2..b73a1dacf 100644 --- a/crates/python/src/py_callable.rs +++ b/crates/python/src/py_callable.rs @@ -1276,7 +1276,7 @@ pub fn wrap_py_llm_stream_exec_intercept_fn( } None => callback.bind(py).call1((py_req, py_next)), } - .map_err(|e: PyErr| FlowError::Internal(e.to_string()))?; + .map_err(python_callback_error)?; let outcome = split_py_object_or_future_with_locals( py, result.unbind(), diff --git a/python/tests/test_llm.py b/python/tests/test_llm.py index 3e3686594..ee64502ef 100644 --- a/python/tests/test_llm.py +++ b/python/tests/test_llm.py @@ -1338,6 +1338,35 @@ async def test_stream_execution_intercept_propagates_direct___anext__error(self) finally: intercepts.deregister_llm_stream_execution("py_llm_stream_direct_error") + async def test_stream_execution_intercept_failure_emits_exception_type(self): + events = [] + subscribers.register("py_llm_stream_intercept_failure_sub", events.append) + + def failing_middleware(request, next): + raise ValueError("stream intercept boom") + + intercepts.register_llm_stream_execution( + "py_llm_stream_failure", + 1, + failing_middleware, + ) + try: + with pytest.raises(RuntimeError, match="stream intercept boom"): + await llm.stream_execute( + "stream_intercept_failure_llm", + make_request(), + lambda request: _single_chunk_stream(), + lambda chunk: None, + lambda: {}, + ) + finally: + intercepts.deregister_llm_stream_execution("py_llm_stream_failure") + subscribers.deregister("py_llm_stream_intercept_failure_sub") + + metadata = _llm_event(events, "stream_intercept_failure_llm", "end").metadata + assert isinstance(metadata, dict) + assert metadata["exception.type"] == "ValueError" + async def test_stream_execute_collector_failure_raises(self): events = [] subscribers.register("py_llm_stream_collector_failure_sub", events.append) From 78e38c05a7cabf8b2021f12a297d69bdf63c9c85 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 15:39:05 -0400 Subject: [PATCH 6/6] test: flush stream events before teardown Signed-off-by: Will Killian --- python/tests/test_llm.py | 1 + 1 file changed, 1 insertion(+) diff --git a/python/tests/test_llm.py b/python/tests/test_llm.py index ee64502ef..ba441981b 100644 --- a/python/tests/test_llm.py +++ b/python/tests/test_llm.py @@ -1361,6 +1361,7 @@ def failing_middleware(request, next): ) finally: intercepts.deregister_llm_stream_execution("py_llm_stream_failure") + await subscribers.flush_async() subscribers.deregister("py_llm_stream_intercept_failure_sub") metadata = _llm_event(events, "stream_intercept_failure_llm", "end").metadata