diff --git a/crates/cli/src/sessions/mod.rs b/crates/cli/src/sessions/mod.rs index 43d66aa0c..77a0b1bf3 100644 --- a/crates/cli/src/sessions/mod.rs +++ b/crates/cli/src/sessions/mod.rs @@ -923,11 +923,22 @@ impl Session { // session-length span. Some session ids can outlive user-visible work, so those harnesses store // metadata here and wait for a bounded turn scope before emitting trace structure. fn start_agent(&mut self, event: SessionEvent) -> Result<(), CliError> { + let emit_start_mark = !self.session_started; self.agent_kind = event.agent_kind; self.session_started = true; self.session_metadata = merge_metadata(self.session_metadata.clone(), event.metadata.clone()); - self.ensure_agent_started(event.metadata) + self.ensure_agent_started(event.metadata.clone())?; + if emit_start_mark { + emit_mark_event( + EmitMarkEventParams::builder() + .name("session.start") + .parent_opt(self.agent_scope.as_ref()) + .metadata(self.scope_metadata(event.metadata)) + .build(), + )?; + } + Ok(()) } // Lazily opens the root agent scope for harnesses that have a meaningful session boundary. @@ -1046,6 +1057,12 @@ impl Session { } fn scope_metadata(&self, event_metadata: Value) -> Value { + let session_instance_id = self + .scope_stack + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .root_uuid() + .to_string(); merge_metadata( merge_metadata( merge_metadata( @@ -1056,6 +1073,7 @@ impl Session { ), json!({ "session_id": self.session_id, + "session_instance_id": session_instance_id, "agent_kind": self.agent_kind.as_str(), "gateway_config_profile": self.config.profile, "plugin_config": self.config.plugin_config, diff --git a/crates/cli/tests/coverage/shared/session_tests.rs b/crates/cli/tests/coverage/shared/session_tests.rs index 028245f96..35f2d14d0 100644 --- a/crates/cli/tests/coverage/shared/session_tests.rs +++ b/crates/cli/tests/coverage/shared/session_tests.rs @@ -265,6 +265,196 @@ fn register_filtered_session_subscriber( .unwrap(); } +#[tokio::test] +async fn session_start_mark_carries_stable_non_overridable_identity_once() { + let subscriber_name = "cli-session-start-identity-test"; + let session_id = "session-start-identity"; + let captured_events = Arc::new(StdMutex::new(Vec::::new())); + let captured = Arc::clone(&captured_events); + register_filtered_session_subscriber( + subscriber_name, + tracked_sessions(&[session_id]), + Arc::new(move |event| captured.lock().unwrap().push(event.clone())), + ); + + let mut config = session_test_config(); + config.metadata = Some(json!({ + "user_id": "alice", + "session_instance_id": "untrusted" + })); + let manager = SessionManager::new(config); + let headers = HeaderMap::new(); + let start = SessionEvent { + session_id: session_id.into(), + agent_kind: AgentKind::Hermes, + event_name: "on_session_start".into(), + payload: json!({"ignored": true}), + metadata: json!({}), + }; + manager + .apply_events( + &headers, + vec![ + NormalizedEvent::AgentStarted(start.clone()), + NormalizedEvent::AgentStarted(start), + NormalizedEvent::AgentEnded(SessionEvent { + session_id: session_id.into(), + agent_kind: AgentKind::Hermes, + event_name: "on_session_end".into(), + payload: json!({}), + metadata: json!({}), + }), + ], + ) + .await + .unwrap(); + flush_subscribers().unwrap(); + + let events = captured_events.lock().unwrap(); + let marks = events + .iter() + .filter(|event| event.scope_category().is_none() && event.name() == "session.start") + .collect::>(); + assert_eq!( + marks.len(), + 1, + "duplicate starts must not duplicate the mark" + ); + let mark = marks[0]; + assert!( + mark.data().is_none(), + "startup hook payload must not be copied" + ); + let metadata = mark.metadata().unwrap(); + assert_eq!(metadata["session_id"], json!(session_id)); + assert_eq!(metadata["user_id"], json!("alice")); + let instance_id = metadata["session_instance_id"].as_str().unwrap(); + let instance_uuid = uuid::Uuid::parse_str(instance_id).unwrap(); + assert_eq!(instance_uuid.get_version(), Some(uuid::Version::SortRand)); + assert_ne!(instance_id, "untrusted"); + + let agent_start = events + .iter() + .find(|event| { + event.scope_category() == Some(ScopeCategory::Start) + && event.scope_type() == Some(ScopeType::Agent) + }) + .unwrap(); + assert_eq!(mark.parent_uuid(), Some(agent_start.uuid())); + assert_eq!( + agent_start.metadata().unwrap()["session_instance_id"], + json!(instance_id) + ); + drop(events); + assert!(deregister_subscriber(subscriber_name).unwrap()); +} + +#[tokio::test] +async fn turn_root_reuses_startup_mark_instance_without_opening_a_turn() { + let subscriber_name = "cli-turn-root-session-instance-test"; + let session_id = "turn-root-session-instance"; + let captured_events = Arc::new(StdMutex::new(Vec::::new())); + let captured = Arc::clone(&captured_events); + register_filtered_session_subscriber( + subscriber_name, + tracked_sessions(&[session_id]), + Arc::new(move |event| captured.lock().unwrap().push(event.clone())), + ); + + let manager = SessionManager::new(session_test_config()); + let headers = HeaderMap::new(); + manager + .apply_events( + &headers, + vec![NormalizedEvent::AgentStarted(codex_session_event( + session_id, + "sessionStart", + json!({"user_id": "alice"}), + ))], + ) + .await + .unwrap(); + { + let sessions = manager.inner.lock().await; + assert!(sessions.get(session_id).unwrap().turn_scope.is_none()); + } + manager + .apply_events( + &headers, + vec![NormalizedEvent::PromptSubmitted(codex_session_event( + session_id, + "UserPromptSubmit", + json!({}), + ))], + ) + .await + .unwrap(); + flush_subscribers().unwrap(); + + let events = captured_events.lock().unwrap(); + let mark = events + .iter() + .find(|event| event.name() == "session.start") + .unwrap(); + let turn_start = events + .iter() + .find(|event| { + event.scope_category() == Some(ScopeCategory::Start) && event.name() == "codex-turn" + }) + .unwrap(); + let instance_id = mark.metadata().unwrap()["session_instance_id"] + .as_str() + .unwrap(); + assert_eq!(mark.parent_uuid(), turn_start.parent_uuid()); + assert_eq!( + turn_start.metadata().unwrap()["session_instance_id"], + json!(instance_id) + ); + drop(events); + assert!(deregister_subscriber(subscriber_name).unwrap()); +} + +#[test] +fn session_instances_are_unique_even_when_logical_session_ids_match() { + let first = Session::new( + "same-session".to_string(), + AgentKind::Codex, + SessionConfig::default(), + ); + let second = Session::new( + "same-session".to_string(), + AgentKind::Codex, + SessionConfig::default(), + ); + let first_root = first.scope_stack.read().unwrap().root_uuid(); + let second_root = second.scope_stack.read().unwrap().root_uuid(); + assert_ne!(first_root, second_root); +} + +#[test] +fn scope_metadata_recovers_poisoned_scope_stack_for_instance_id() { + let session = Session::new( + "poisoned-session".to_string(), + AgentKind::Codex, + SessionConfig::default(), + ); + let root_uuid = session.scope_stack.read().unwrap().root_uuid(); + let scope_stack = session.scope_stack.clone(); + std::thread::spawn(move || { + let _guard = scope_stack.write().unwrap(); + panic!("poison session scope stack for recovery test"); + }) + .join() + .expect_err("fixture scope stack writer should panic"); + + let metadata = session.scope_metadata(Value::Null); + + assert_eq!( + metadata["session_instance_id"], + json!(root_uuid.to_string()) + ); +} + async fn apply_codex_payload(manager: &SessionManager, headers: &HeaderMap, payload: Value) { let outcome = crate::agents::shared::adapters::codex::adapt(payload, headers); manager.apply_events(headers, outcome.events).await.unwrap(); @@ -1984,7 +2174,7 @@ async fn writes_atif_on_session_end_from_plugin_config() { let mut headers = HeaderMap::new(); headers.insert( "x-nemo-relay-session-metadata", - r#"{"team":"coverage"}"#.parse().unwrap(), + r#"{"team":"coverage","user_id":"alice"}"#.parse().unwrap(), ); headers.insert("x-nemo-relay-gateway-mode", "required".parse().unwrap()); @@ -2029,6 +2219,18 @@ async fn writes_atif_on_session_end_from_plugin_config() { atif["extra"]["observed_events"][0]["name"], json!("codex-turn") ); + assert_eq!( + atif["extra"]["nemo_relay"]["session_id"], + json!("atif-session") + ); + assert_eq!(atif["extra"]["nemo_relay"]["user_id"], json!("alice")); + let session_instance_id = atif["extra"]["nemo_relay"]["session_instance_id"] + .as_str() + .unwrap(); + assert_eq!( + atif["extra"]["observed_events"][0]["metadata"]["session_instance_id"], + json!(session_instance_id) + ); } #[tokio::test] diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index c3c6bf74c..f04e71cff 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -117,6 +117,36 @@ pub(crate) fn push_top_level_json_attributes( } } +/// Adds canonical session-correlation attributes from event metadata and the +/// active scope-stack instance. +#[cfg(any(feature = "otel", feature = "openinference"))] +pub(crate) fn push_session_identity_attributes( + attributes: &mut Vec, + event: &crate::api::event::Event, +) { + use opentelemetry::KeyValue; + + let metadata = event.metadata(); + if let Some(session_id) = metadata + .and_then(|value| value.get("session_id")) + .and_then(crate::json::Json::as_str) + { + attributes.push(KeyValue::new("session.id", session_id.to_string())); + } + if let Some(user_id) = metadata + .and_then(|value| value.get("user_id")) + .and_then(crate::json::Json::as_str) + { + attributes.push(KeyValue::new("user.id", user_id.to_string())); + } + if let Ok(stack) = crate::api::runtime::current_scope_stack().read() { + attributes.push(KeyValue::new( + "nemo_relay.session.instance_id", + stack.root_uuid().to_string(), + )); + } +} + #[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( diff --git a/crates/core/src/observability/openinference.rs b/crates/core/src/observability/openinference.rs index 4e3d9c1a0..4171155c2 100644 --- a/crates/core/src/observability/openinference.rs +++ b/crates/core/src/observability/openinference.rs @@ -25,7 +25,7 @@ use super::{ 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, push_serialized_top_level_attributes, - push_top_level_json_attributes, validate_attribute_mappings, + push_session_identity_attributes, push_top_level_json_attributes, validate_attribute_mappings, }; use crate::api::event::{Event, EventNormalizationExt, ScopeCategory}; use crate::api::runtime::EventSubscriberFn; @@ -627,13 +627,18 @@ impl OpenInferenceEventProcessor { fn process_start(&mut self, event: &Event) { self.remove_completed_span_context(event.uuid()); + let parent_context = self.parent_context(event); + let is_trace_root = !parent_context.span().span_context().is_valid(); let mut span = self .tracer .span_builder(span_name(event)) .with_kind(span_kind(event)) .with_start_time(to_system_time(*event.timestamp())) - .start_with_context(&self.tracer, &self.parent_context(event)); - let attributes = start_attributes(event); + .start_with_context(&self.tracer, &parent_context); + let mut attributes = start_attributes(event); + if is_trace_root { + push_session_identity_attributes(&mut 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()); @@ -678,6 +683,9 @@ impl OpenInferenceEventProcessor { let mark_name = event.name().to_string(); let timestamp = to_system_time(*event.timestamp()); let mut attributes = mark_attributes(event); + if event.name() == "session.start" { + push_session_identity_attributes(&mut attributes, event); + } if self.find_parent_span(event).is_some() { apply_attribute_mappings(&mut attributes, &self.attribute_mappings); @@ -710,6 +718,9 @@ impl OpenInferenceEventProcessor { let timestamp = to_system_time(*event.timestamp()); let orphan = self.find_parent_span(event).is_none(); let mut attributes = mark_attributes(event); + if event.name() == "session.start" { + push_session_identity_attributes(&mut attributes, event); + } attributes.push(KeyValue::new("nemo_relay.mark.projection", "tool")); attributes.push(KeyValue::new( oi::OPENINFERENCE_SPAN_KIND, diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 671367036..9610d81f6 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -24,8 +24,8 @@ use super::{ 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, push_serialized_top_level_attributes, push_top_level_json_attributes, - validate_attribute_mappings, + model_name_for_llm_event, push_serialized_top_level_attributes, + push_session_identity_attributes, push_top_level_json_attributes, validate_attribute_mappings, }; use crate::api::event::{Event, EventNormalizationExt, ScopeCategory}; use crate::api::runtime::EventSubscriberFn; @@ -620,13 +620,18 @@ impl OtelEventProcessor { fn process_start(&mut self, event: &Event) { self.remove_completed_span_context(event.uuid()); + let parent_context = self.parent_context(event); + let is_trace_root = !parent_context.span().span_context().is_valid(); let mut span = self .tracer .span_builder(span_name(event)) .with_kind(span_kind(event)) .with_start_time(to_system_time(*event.timestamp())) - .start_with_context(&self.tracer, &self.parent_context(event)); - let attributes = start_attributes(event); + .start_with_context(&self.tracer, &parent_context); + let mut attributes = start_attributes(event); + if is_trace_root { + push_session_identity_attributes(&mut 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()); @@ -672,6 +677,9 @@ impl OtelEventProcessor { let mark_name = event.name().to_string(); let timestamp = to_system_time(*event.timestamp()); let mut attributes = mark_attributes(event); + if event.name() == "session.start" { + push_session_identity_attributes(&mut attributes, event); + } if self.find_parent_span(event).is_some() { apply_attribute_mappings(&mut attributes, &self.attribute_mappings); @@ -700,6 +708,9 @@ impl OtelEventProcessor { let timestamp = to_system_time(*event.timestamp()); let orphan = self.find_parent_span(event).is_none(); let mut attributes = mark_attributes(event); + if event.name() == "session.start" { + push_session_identity_attributes(&mut attributes, event); + } attributes.push(KeyValue::new("nemo_relay.mark.projection", "tool")); attributes.push(KeyValue::new("nemo_relay.scope_type", "tool")); if orphan { diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 94f025390..e8d39b930 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -1005,6 +1005,7 @@ struct ManagedAtifExporter { exporter: AtifExporter, filename: String, local_path: Option, + correlation: AtifCorrelation, observed_events: Vec, observed_event_keys: HashSet, written: bool, @@ -1032,6 +1033,51 @@ struct PendingAtifExport { exporter: AtifExporter, filename: String, local_path: Option, + correlation: AtifCorrelation, +} + +#[derive(Clone)] +struct AtifCorrelation { + session_id: Option, + session_instance_id: Option, + user_id: Option, +} + +impl AtifCorrelation { + fn from_event(event: &Event) -> Self { + let metadata = event.metadata(); + Self { + session_id: metadata + .and_then(|value| value.get("session_id")) + .and_then(Json::as_str) + .map(ToString::to_string), + session_instance_id: current_scope_stack() + .read() + .ok() + .map(|stack| stack.root_uuid().to_string()), + user_id: metadata + .and_then(|value| value.get("user_id")) + .and_then(Json::as_str) + .map(ToString::to_string), + } + } + + fn to_json(&self) -> Json { + let mut fields = Map::new(); + if let Some(session_id) = &self.session_id { + fields.insert("session_id".to_string(), Json::String(session_id.clone())); + } + if let Some(session_instance_id) = &self.session_instance_id { + fields.insert( + "session_instance_id".to_string(), + Json::String(session_instance_id.clone()), + ); + } + if let Some(user_id) = &self.user_id { + fields.insert("user_id".to_string(), Json::String(user_id.clone())); + } + Json::Object(fields) + } } /// Identifier for a single output sink. `Local` is used when `storage` is empty @@ -1082,6 +1128,7 @@ impl AtifDispatcher { let exporter = AtifExporter::new(session_id.clone(), self.agent_info()); (exporter.subscriber())(event); let (filename, local_path) = self.prepare_destination(&session_id); + let correlation = AtifCorrelation::from_event(event); self.scope_owners.insert(event.uuid(), event.uuid()); self.agents.insert( event.uuid(), @@ -1089,6 +1136,7 @@ impl AtifDispatcher { exporter, filename, local_path, + correlation, observed_events: vec![event.clone()], observed_event_keys: HashSet::from([event_observation_key(event)]), written: false, @@ -1206,6 +1254,7 @@ impl AtifDispatcher { exporter: agent.exporter.clone(), filename: agent.filename.clone(), local_path: agent.local_path.clone(), + correlation: agent.correlation.clone(), }); } } @@ -1349,6 +1398,7 @@ fn prepare_atif_file( agent.local_path.clone(), trajectory, observed_events, + agent.correlation.clone(), ) } @@ -1372,6 +1422,7 @@ fn prepare_atif_shutdown_file( export.local_path.clone(), trajectory, observed_events, + export.correlation.clone(), ) } @@ -1381,15 +1432,30 @@ fn prepare_atif_payload( local_path: Option, trajectory: crate::observability::atif::AtifTrajectory, observed_events: Vec, + correlation: AtifCorrelation, ) -> std::io::Result { let mut value = serde_json::to_value(trajectory)?; if let Some(object) = value.as_object_mut() { - object.insert( - "extra".to_string(), - serde_json::json!({ - "observed_events": observed_events, - }), + let existing_extra = object.remove("extra"); + let mut extra = match existing_extra { + Some(Json::Object(fields)) => fields, + Some(value) => Map::from_iter([("trajectory_extra".to_string(), value)]), + None => Map::new(), + }; + extra.insert( + "observed_events".to_string(), + serde_json::to_value(observed_events)?, ); + let mut nemo_relay = match extra.remove("nemo_relay") { + Some(Json::Object(fields)) => fields, + Some(value) => Map::from_iter([("trajectory_extra".to_string(), value)]), + None => Map::new(), + }; + if let Json::Object(correlation_fields) = correlation.to_json() { + nemo_relay.extend(correlation_fields); + } + extra.insert("nemo_relay".to_string(), Json::Object(nemo_relay)); + object.insert("extra".to_string(), Json::Object(extra)); } let payload = serde_json::to_vec_pretty(&value)?; Ok(PendingAtifWrite { diff --git a/crates/core/tests/unit/observability/exporter_parity_tests.rs b/crates/core/tests/unit/observability/exporter_parity_tests.rs index 4ea93683a..ea40a4f19 100644 --- a/crates/core/tests/unit/observability/exporter_parity_tests.rs +++ b/crates/core/tests/unit/observability/exporter_parity_tests.rs @@ -20,6 +20,7 @@ use crate::api::event::{ BaseEvent, CategoryProfile, Event, EventCategory, EventNormalizationExt, ScopeCategory, ScopeEvent, }; +use crate::api::runtime::{create_scope_stack, with_scope_stack}; use crate::codec::model_pricing::pricing_test_mutex; use crate::codec::response::{ PricingCatalog, PricingResolver, reset_active_pricing_resolver, set_active_pricing_resolver, @@ -816,3 +817,65 @@ fn test_manual_fallback_payload_parity() { Some(&"1500".to_string()) ); } + +#[test] +fn session_instance_correlates_distinct_otel_and_openinference_traces() { + let scope_stack = create_scope_stack(); + let expected_root_uuid = scope_stack.read().unwrap().root_uuid().to_string(); + let uuid = Uuid::now_v7(); + let metadata = json!({"session_id": "logical-session", "user_id": "alice"}); + let start = Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .uuid(uuid) + .name("session-root") + .metadata(metadata) + .build(), + ScopeCategory::Start, + Vec::new(), + EventCategory::agent(), + None, + )); + let end = make_scope_event( + ScopeCategory::End, + uuid, + "session-root", + EventCategory::agent(), + None, + None, + ); + let exports = with_scope_stack(scope_stack, || export_through_all_exporters(&[start, end])); + let otel = exports + .otel_spans + .iter() + .find(|span| span.name.as_ref() == "session-root") + .unwrap(); + let openinference = exports + .openinference_spans + .iter() + .find(|span| span.name.as_ref() == "session-root") + .unwrap(); + let otel_attributes = attr_map(&otel.attributes); + let openinference_attributes = attr_map(&openinference.attributes); + + assert_eq!(otel_attributes["nemo_relay.uuid"], uuid.to_string()); + assert_eq!( + openinference_attributes["nemo_relay.uuid"], + uuid.to_string() + ); + assert_eq!(otel_attributes["session.id"], "logical-session"); + assert_eq!(openinference_attributes["session.id"], "logical-session"); + assert_eq!(otel_attributes["user.id"], "alice"); + assert_eq!(openinference_attributes["user.id"], "alice"); + assert_eq!( + otel_attributes["nemo_relay.session.instance_id"], + expected_root_uuid + ); + assert_eq!( + openinference_attributes["nemo_relay.session.instance_id"], + expected_root_uuid + ); + assert_ne!( + otel.span_context.trace_id(), + openinference.span_context.trace_id() + ); +} diff --git a/crates/core/tests/unit/observability/openinference_tests.rs b/crates/core/tests/unit/observability/openinference_tests.rs index 7f5a03c12..347617bd4 100644 --- a/crates/core/tests/unit/observability/openinference_tests.rs +++ b/crates/core/tests/unit/observability/openinference_tests.rs @@ -427,6 +427,26 @@ fn make_start_event( ) } +fn make_start_event_with_metadata( + uuid: Uuid, + parent_uuid: Option, + name: &str, + metadata: Json, +) -> Event { + Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .parent_uuid_opt(parent_uuid) + .uuid(uuid) + .name(name) + .metadata(metadata) + .build(), + ScopeCategory::Start, + Vec::new(), + EventCategory::agent(), + None, + )) +} + fn make_end_event( uuid: Uuid, parent_uuid: Option, @@ -524,6 +544,18 @@ fn make_mark_event(parent_uuid: Option, name: &str, data: Option) -> )) } +fn make_mark_event_with_metadata(parent_uuid: Option, metadata: Json) -> Event { + Event::Mark(MarkEvent::new( + BaseEvent::builder() + .parent_uuid_opt(parent_uuid) + .name("session.start") + .metadata(metadata) + .build(), + None, + None, + )) +} + struct CapturedHttpRequest { path: String, content_type: String, @@ -773,6 +805,136 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { ); } +#[test] +fn session_identity_is_projected_on_trace_roots_and_marks_only() { + let (provider, exporter) = make_provider(); + let subscriber = OpenInferenceSubscriber::from_tracer_provider(provider, "session-identity"); + let callback = subscriber.subscriber(); + let root_uuid = Uuid::now_v7(); + let child_uuid = Uuid::now_v7(); + let second_root_uuid = Uuid::now_v7(); + let instance_id = crate::api::runtime::current_scope_stack() + .read() + .unwrap() + .root_uuid() + .to_string(); + let identity = json!({"session_id": "logical-session", "user_id": "alice"}); + + callback(&make_start_event_with_metadata( + root_uuid, + None, + "identity-root", + identity.clone(), + )); + callback(&make_start_event_with_metadata( + child_uuid, + Some(root_uuid), + "identity-child", + identity.clone(), + )); + callback(&make_mark_event_with_metadata( + Some(root_uuid), + identity.clone(), + )); + callback(&make_end_event( + child_uuid, + Some(root_uuid), + "identity-child", + ScopeType::Agent, + None, + )); + callback(&make_end_event( + root_uuid, + None, + "identity-root", + ScopeType::Agent, + None, + )); + callback(&make_start_event_with_metadata( + second_root_uuid, + None, + "identity-second-root", + json!({"session_id": "logical-session", "user_id": 42}), + )); + callback(&make_end_event( + second_root_uuid, + None, + "identity-second-root", + ScopeType::Agent, + None, + )); + callback(&make_mark_event_with_metadata(None, identity)); + subscriber.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + let root = spans + .iter() + .find(|span| span.name.as_ref() == "identity-root") + .unwrap(); + let child = spans + .iter() + .find(|span| span.name.as_ref() == "identity-child") + .unwrap(); + let second_root = spans + .iter() + .find(|span| span.name.as_ref() == "identity-second-root") + .unwrap(); + let orphan_mark = spans + .iter() + .find(|span| span.name.as_ref() == "mark:session.start") + .unwrap(); + + let root_attributes = attr_map(&root.attributes); + assert_eq!(root_attributes["session.id"], "logical-session"); + assert_eq!(root_attributes["user.id"], "alice"); + assert_eq!( + root_attributes["nemo_relay.session.instance_id"], + instance_id + ); + assert_eq!( + root.attributes + .iter() + .filter(|attribute| attribute.key.as_str() == "session.id") + .count(), + 1 + ); + let child_attributes = attr_map(&child.attributes); + assert!(!child_attributes.contains_key("session.id")); + assert!(!child_attributes.contains_key("user.id")); + assert!(!child_attributes.contains_key("nemo_relay.session.instance_id")); + assert_eq!(child_attributes["openinference.metadata.user_id"], "alice"); + let second_root_attributes = attr_map(&second_root.attributes); + assert_eq!(second_root_attributes["session.id"], "logical-session"); + assert!(!second_root_attributes.contains_key("user.id")); + assert_eq!( + second_root_attributes["openinference.metadata.user_id"], + "42" + ); + assert_eq!( + second_root_attributes["nemo_relay.session.instance_id"], + root_attributes["nemo_relay.session.instance_id"] + ); + assert_ne!( + root.span_context.trace_id(), + second_root.span_context.trace_id() + ); + + let mark_attributes = attr_map(&root.events.events[0].attributes); + assert_eq!(mark_attributes["session.id"], "logical-session"); + assert_eq!(mark_attributes["user.id"], "alice"); + assert_eq!( + mark_attributes["nemo_relay.session.instance_id"], + root_attributes["nemo_relay.session.instance_id"] + ); + let orphan_attributes = attr_map(&orphan_mark.attributes); + assert_eq!(orphan_attributes["session.id"], "logical-session"); + assert_eq!(orphan_attributes["user.id"], "alice"); + assert_eq!( + orphan_attributes["nemo_relay.session.instance_id"], + root_attributes["nemo_relay.session.instance_id"] + ); +} + #[test] fn mapped_orphan_mark_alias_cannot_replace_intrinsic_mark_fields() { let (provider, exporter) = make_provider(); diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index a5b790ccb..2422e43a9 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -329,6 +329,26 @@ fn make_start_event( ) } +fn make_start_event_with_metadata( + uuid: Uuid, + parent_uuid: Option, + name: &str, + metadata: Json, +) -> Event { + Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .parent_uuid_opt(parent_uuid) + .uuid(uuid) + .name(name) + .metadata(metadata) + .build(), + ScopeCategory::Start, + Vec::new(), + EventCategory::agent(), + None, + )) +} + fn make_end_event( uuid: Uuid, parent_uuid: Option, @@ -426,6 +446,18 @@ fn make_mark_event(parent_uuid: Option, name: &str, data: Option) -> )) } +fn make_mark_event_with_metadata(parent_uuid: Option, metadata: Json) -> Event { + Event::Mark(MarkEvent::new( + BaseEvent::builder() + .parent_uuid_opt(parent_uuid) + .name("session.start") + .metadata(metadata) + .build(), + None, + None, + )) +} + struct CapturedHttpRequest { path: String, content_type: String, @@ -670,6 +702,139 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { ); } +#[test] +fn session_identity_is_projected_on_trace_roots_and_marks_only() { + let (provider, exporter) = make_provider(); + let subscriber = OpenTelemetrySubscriber::from_tracer_provider(provider, "session-identity"); + let callback = subscriber.subscriber(); + let root_uuid = Uuid::now_v7(); + let child_uuid = Uuid::now_v7(); + let second_root_uuid = Uuid::now_v7(); + let instance_id = crate::api::runtime::current_scope_stack() + .read() + .unwrap() + .root_uuid() + .to_string(); + let identity = json!({"session_id": "logical-session", "user_id": "alice"}); + + callback(&make_start_event_with_metadata( + root_uuid, + None, + "identity-root", + identity.clone(), + )); + callback(&make_start_event_with_metadata( + child_uuid, + Some(root_uuid), + "identity-child", + identity.clone(), + )); + callback(&make_mark_event_with_metadata( + Some(root_uuid), + identity.clone(), + )); + callback(&make_end_event( + child_uuid, + Some(root_uuid), + "identity-child", + ScopeType::Agent, + None, + )); + callback(&make_end_event( + root_uuid, + None, + "identity-root", + ScopeType::Agent, + None, + )); + callback(&make_start_event_with_metadata( + second_root_uuid, + None, + "identity-second-root", + json!({"session_id": "logical-session", "user_id": 42}), + )); + callback(&make_end_event( + second_root_uuid, + None, + "identity-second-root", + ScopeType::Agent, + None, + )); + callback(&make_mark_event_with_metadata(None, identity)); + subscriber.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + let root = spans + .iter() + .find(|span| span.name.as_ref() == "identity-root") + .unwrap(); + let child = spans + .iter() + .find(|span| span.name.as_ref() == "identity-child") + .unwrap(); + let second_root = spans + .iter() + .find(|span| span.name.as_ref() == "identity-second-root") + .unwrap(); + let orphan_mark = spans + .iter() + .find(|span| span.name.as_ref() == "mark:session.start") + .unwrap(); + + let root_attributes = attr_map(&root.attributes); + assert_eq!(root_attributes["session.id"], "logical-session"); + assert_eq!(root_attributes["user.id"], "alice"); + assert_eq!( + root_attributes["nemo_relay.session.instance_id"], + instance_id + ); + assert_eq!( + root.attributes + .iter() + .filter(|attribute| attribute.key.as_str() == "session.id") + .count(), + 1 + ); + let child_attributes = attr_map(&child.attributes); + assert!(!child_attributes.contains_key("session.id")); + assert!(!child_attributes.contains_key("user.id")); + assert!(!child_attributes.contains_key("nemo_relay.session.instance_id")); + assert_eq!( + child_attributes["nemo_relay.start.metadata.user_id"], + "alice" + ); + let second_root_attributes = attr_map(&second_root.attributes); + assert_eq!(second_root_attributes["session.id"], "logical-session"); + assert!(!second_root_attributes.contains_key("user.id")); + assert_eq!( + second_root_attributes["nemo_relay.start.metadata.user_id"], + "42" + ); + assert_eq!( + second_root_attributes["nemo_relay.session.instance_id"], + root_attributes["nemo_relay.session.instance_id"] + ); + assert_ne!( + root.span_context.trace_id(), + second_root.span_context.trace_id() + ); + + let mark_attributes = attr_map(&root.events.events[0].attributes); + assert_eq!(mark_attributes["session.id"], "logical-session"); + assert_eq!(mark_attributes["user.id"], "alice"); + assert_eq!( + mark_attributes["nemo_relay.session.instance_id"], + root_attributes["nemo_relay.session.instance_id"] + ); + let orphan_attributes = attr_map(&orphan_mark.attributes); + assert_eq!(orphan_attributes["session.id"], "logical-session"); + assert_eq!(orphan_attributes["user.id"], "alice"); + assert_eq!( + orphan_attributes["nemo_relay.session.instance_id"], + root_attributes["nemo_relay.session.instance_id"] + ); +} + #[test] fn mapped_orphan_mark_alias_cannot_replace_intrinsic_mark_fields() { let (provider, exporter) = make_provider(); diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 419f32121..abddf6e89 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -1085,7 +1085,10 @@ fn atif_routes_global_descendant_events_by_parent_uuid() { .uuid(agent_uuid) .parent_uuid(root_uuid) .name("root-agent") - .metadata(json!({"session_id": "root-session"})) + .metadata(json!({ + "session_id": "root-session", + "user_id": "alice" + })) .build(), ScopeCategory::Start, vec![], @@ -1105,6 +1108,32 @@ fn atif_routes_global_descendant_events_by_parent_uuid() { .is_none() ); + let session_start_mark = Event::Mark(MarkEvent::new( + BaseEvent::builder() + .parent_uuid(agent_uuid) + .name("session.start") + .metadata(json!({ + "session_id": "root-session", + "user_id": "alice", + "session_instance_id": root_uuid.to_string() + })) + .build(), + None, + None, + )); + assert!( + manager + .lock() + .unwrap() + .observe_global( + &session_start_mark, + "__test__", + Arc::clone(&manager), + Arc::clone(&empty_storage), + ) + .is_none() + ); + let child_start_event = Event::Scope(ScopeEvent::new( BaseEvent::builder() .uuid(child_uuid) @@ -1190,6 +1219,27 @@ fn atif_routes_global_descendant_events_by_parent_uuid() { let value: Json = serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap(); assert_eq!(value["trajectory_id"], agent_uuid.to_string()); + assert_eq!(value["session_id"], agent_uuid.to_string()); + assert_eq!(value["extra"]["nemo_relay"]["session_id"], "root-session"); + assert_eq!( + value["extra"]["nemo_relay"]["session_instance_id"], + root_uuid.to_string() + ); + assert_eq!(value["extra"]["nemo_relay"]["user_id"], "alice"); + assert_eq!( + value["extra"]["observed_events"] + .as_array() + .unwrap() + .iter() + .filter(|event| event["name"] == "session.start") + .count(), + 1 + ); + assert!( + value["steps"] + .as_array() + .is_some_and(|steps| steps.iter().all(|step| step["event"] != "session.start")) + ); assert_eq!( value["subagent_trajectories"][0]["session_id"], "child-session" @@ -1491,6 +1541,64 @@ fn atif_dispatcher_default_output_path_uses_current_directory() { ); } +#[test] +fn atif_payload_merges_correlation_with_existing_trajectory_extra() { + let agent_uuid = Uuid::now_v7(); + let trajectory = crate::observability::atif::AtifTrajectory { + schema_version: "ATIF-v1.7".to_string(), + session_id: agent_uuid.to_string(), + trajectory_id: Some(agent_uuid.to_string()), + agent: AtifAgentInfo { + name: "test-agent".to_string(), + version: "1.0.0".to_string(), + model_name: None, + tool_definitions: None, + extra: None, + }, + steps: Vec::new(), + notes: None, + final_metrics: None, + continued_trajectory_ref: None, + subagent_trajectories: None, + extra: Some(json!({ + "existing": "preserved", + "nemo_relay": { + "existing": "nested", + "session_id": "untrusted" + } + })), + }; + let write = prepare_atif_payload( + agent_uuid, + format!("trajectory-{agent_uuid}.json"), + None, + trajectory, + Vec::new(), + AtifCorrelation { + session_id: Some("logical-session".to_string()), + session_instance_id: Some("instance-id".to_string()), + user_id: Some("alice".to_string()), + }, + ) + .unwrap(); + let value: Json = serde_json::from_slice(&write.payload).unwrap(); + + assert_eq!(value["session_id"], agent_uuid.to_string()); + assert_eq!(value["trajectory_id"], agent_uuid.to_string()); + assert_eq!(value["extra"]["existing"], "preserved"); + assert_eq!(value["extra"]["nemo_relay"]["existing"], "nested"); + assert_eq!( + value["extra"]["nemo_relay"]["session_id"], + "logical-session" + ); + assert_eq!( + value["extra"]["nemo_relay"]["session_instance_id"], + "instance-id" + ); + assert_eq!(value["extra"]["nemo_relay"]["user_id"], "alice"); + assert_eq!(value["extra"]["observed_events"], json!([])); +} + #[test] fn atif_explicit_options_and_open_agent_teardown_are_written() { let _guard = crate::observability::test_mutex().lock().unwrap(); diff --git a/docs/about-nemo-relay/concepts/events.mdx b/docs/about-nemo-relay/concepts/events.mdx index dddf27cd0..537a1175e 100644 --- a/docs/about-nemo-relay/concepts/events.mdx +++ b/docs/about-nemo-relay/concepts/events.mdx @@ -32,6 +32,12 @@ Emitted when lifecycle work starts or ends. `scope_category` is `start` or Emitted for named runtime checkpoints that are not full start/end lifecycle pairs. +The coding-agent CLI emits one `session.start` mark when a session lifecycle +begins. Its metadata includes the logical harness `session_id`, the Relay-owned +`session_instance_id`, and any configured session metadata such as `user_id`. +The mark does not copy the native startup-hook payload or force a turn scope to +open. + ## Shared Envelope Every event carries a shared envelope: diff --git a/docs/configure-plugins/observability/about.mdx b/docs/configure-plugins/observability/about.mdx index 30d2bc802..85ef99e54 100644 --- a/docs/configure-plugins/observability/about.mdx +++ b/docs/configure-plugins/observability/about.mdx @@ -123,6 +123,15 @@ OpenTelemetry and OpenInference spans carry the same values as `trace_id` and `span_id` values are still generated by the tracing backend and are not written into ATIF. +For coding-agent sessions, OTLP trace roots additionally carry the logical +`session.id`, optional `user.id`, and Relay-owned +`nemo_relay.session.instance_id`. The instance value is the existing root scope +UUID and is shared across trace roots created by the same runtime session. +OpenTelemetry and OpenInference generate independent native trace IDs, so use +the Relay instance ID for cross-exporter or cross-trace correlation and use +each backend's `trace_id` to join child rows. ATIF keeps its existing artifact +IDs and exposes the same logical values under `extra.nemo_relay`. + ## Pages - [Observability Configuration](/configure-plugins/observability/configuration) documents the whole plugin diff --git a/docs/configure-plugins/observability/atif.mdx b/docs/configure-plugins/observability/atif.mdx index d56617c69..1063f1d33 100644 --- a/docs/configure-plugins/observability/atif.mdx +++ b/docs/configure-plugins/observability/atif.mdx @@ -45,8 +45,15 @@ coding-agent turn scope. ATIF contains agent interaction steps, tool calls, and observations. Relay mark events are point-in-time telemetry rather than trajectory steps, so the ATIF -exporter omits them. Use ATOF to retain the canonical mark event stream, or an -OTLP exporter when marks must remain correlated with trace data. +exporter omits them from `steps`. Plugin-managed files retain raw events that +were associated with the trajectory under `extra.observed_events`. + +Plugin-managed files also expose selected correlation fields under +`extra.nemo_relay`: `session_id` is the logical harness session ID, +`session_instance_id` is Relay's root scope UUID, and `user_id` is present when +session metadata contains a top-level string value. The top-level ATIF +`session_id`, `trajectory_id`, `{session_id}` filename expansion, and remote +session header remain trajectory-unique artifact identities for compatibility. ## Fields diff --git a/docs/configure-plugins/observability/atof.mdx b/docs/configure-plugins/observability/atof.mdx index d0103354e..96a4def45 100644 --- a/docs/configure-plugins/observability/atof.mdx +++ b/docs/configure-plugins/observability/atof.mdx @@ -99,6 +99,12 @@ Each emitted scope, tool, LLM, middleware, or mark event is written as one ATOF JSON object per line. For event field semantics, refer to [Events](/about-nemo-relay/concepts/events). +Coding-agent sessions emit a `session.start` mark with the logical +`metadata.session_id`, a Relay-owned UUIDv7 `metadata.session_instance_id`, and +optional session metadata such as `metadata.user_id`. Later trace exporters +use these identities for correlation, but ATOF preserves them as raw event +metadata. + ATOF is the raw-event export path. It preserves the canonical event shape; it does not add semantic extraction, replay policy, scope-owner resolution, or exporter projection rules. diff --git a/docs/configure-plugins/observability/openinference.mdx b/docs/configure-plugins/observability/openinference.mdx index 423d5435f..2aeff3b3d 100644 --- a/docs/configure-plugins/observability/openinference.mdx +++ b/docs/configure-plugins/observability/openinference.mdx @@ -106,6 +106,16 @@ attributes. These values match ATIF `step.extra.ancestry.function_id` and the trajectory-root span's `nemo_relay.uuid` also matches the ATIF `session_id`. Backend-native `trace_id` and `span_id` values are not written into ATIF. +Coding-agent trace roots also carry `session.id`, optional `user.id`, and +`nemo_relay.session.instance_id`. The first value is the logical harness +session ID, while the Relay instance ID is the existing root scope UUID shared +by trace roots from one runtime session instance. These fields are root-only: +filter matching roots first, then use the OpenInference backend's `trace_id` to +select child rows. OpenInference and generic OpenTelemetry export generate +independent trace and span IDs, but retain the same Relay instance ID and +`nemo_relay.uuid` values. The `session.start` mark carries the same correlation +fields whether represented as a span event or an orphan zero-duration span. + LLM token counts appear as `llm.token_count.prompt`, `llm.token_count.completion`, `llm.token_count.total`, and `llm.token_count.prompt_details.cache_read`/`cache_write`. Cost appears as USD-denominated `llm.cost.total`. Refer to diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index 126c689cd..bc31c68c2 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -94,6 +94,21 @@ attributes. These values match ATIF `step.extra.ancestry.function_id` and the trajectory-root span's `nemo_relay.uuid` also matches the ATIF `session_id`. Backend-native `trace_id` and `span_id` values are not written into ATIF. +Coding-agent trace roots also carry these correlation attributes: + +| Attribute | Meaning | +|---|---| +| `session.id` | Logical harness session ID. | +| `user.id` | Optional top-level string `user_id` from session metadata. | +| `nemo_relay.session.instance_id` | Relay-owned root scope UUID shared by trace roots from the same runtime session instance. | + +These canonical fields are emitted on trace roots rather than repeated on +every child span. Filter roots by `session.id`, `user.id`, or +`nemo_relay.session.instance_id`, then use the backend `trace_id` to select the +remaining rows in each trace. A `session.start` mark carries the same fields: +it is a span event when a parent is active and a zero-duration root span when +it is orphaned. + For LLM end spans, cost is emitted as `nemo_relay.llm.cost.total` and `nemo_relay.llm.cost.currency` (any currency). Token counts are not emitted as discrete attributes. Refer to diff --git a/docs/nemo-relay-cli/basic-usage.mdx b/docs/nemo-relay-cli/basic-usage.mdx index 674ebed8b..d49c14d27 100644 --- a/docs/nemo-relay-cli/basic-usage.mdx +++ b/docs/nemo-relay-cli/basic-usage.mdx @@ -241,6 +241,18 @@ Plugin configuration controls process-level Observability exporters. Per-session configuration controls structured metadata on the top-level agent begin event and the plugin configuration metadata associated with the session. +For example, attach a user identity to a transparent run: + +```bash +nemo-relay run --session-metadata '{"user_id":"alice"}' -- codex +``` + +The CLI emits one `session.start` mark for the lifecycle and copies the merged +metadata onto each top-level trace scope. It also adds a non-overridable +`session_instance_id` derived from the session's existing root scope UUID. +Keep `user_id` a top-level string when OTLP exporters should expose it as +`user.id`. + `hook-forward` can also pass per-session configuration through headers: - `x-nemo-relay-config-profile` @@ -258,8 +270,9 @@ where provider traffic was expected to pass through the gateway. The gateway normalizes vendor hook payloads into private internal events before calling NeMo Relay APIs. -- Agent start opens a top-level `ScopeType::Agent` scope on a dedicated - `ScopeStackHandle`. +- Agent start emits a `session.start` mark on a dedicated `ScopeStackHandle`. + Harnesses with a reliable session boundary also open a top-level + `ScopeType::Agent` scope; Claude Code and Codex use bounded turn scopes. - Subagent start opens a child `ScopeType::Agent` scope. Subagent stop closes that scope when it is still active. - Tool pre-use starts a NeMo Relay tool span. Tool post-use, denial, or failure diff --git a/integrations/coding-agents/README.md b/integrations/coding-agents/README.md index 9a9c09a0c..ba11361cd 100644 --- a/integrations/coding-agents/README.md +++ b/integrations/coding-agents/README.md @@ -192,7 +192,8 @@ status to the host. Useful wrapper options: - `--session-metadata ''` adds structured metadata to the agent begin - event. + event. For example, `--session-metadata '{"user_id":"alice"}'` exposes the + string as `user.id` on OTLP trace roots. - `--profile ` records a configuration profile in session metadata. - `--gateway-mode hook-only|passthrough|required` records the expected gateway behavior in session metadata.