feat(observability): correlate startup sessions across exports

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,

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::<Event>::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::<Vec<_>>();
+    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::<Event>::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]

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<opentelemetry::KeyValue>,
+    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<T: Serialize + ?Sized>(

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,