diff --git a/crates/core/src/observability/atif.rs b/crates/core/src/observability/atif.rs index ce96140e9..0e52d80e1 100644 --- a/crates/core/src/observability/atif.rs +++ b/crates/core/src/observability/atif.rs @@ -22,7 +22,7 @@ //! | LLM End | `agent` step | Response content, tool_calls promoted| //! | Tool Start | *(skipped)* | tool_calls come from LLM End instead | //! | Tool End | agent observation | Correlated by `source_call_id` | -//! | Mark (with data)| `system` step | Custom event data preserved | +//! | Mark | `system` step | Name, payload, metadata, category, subtype preserved in `extra` | //! | Scope Start/End | *(skipped)* | Structural events, not trajectory | //! //! The exporter serializes the full collected event stream into a single ATIF @@ -276,6 +276,15 @@ pub struct AtifStepExtra { /// Full raw point-in-time event payload for mark/system steps. #[serde(skip_serializing_if = "Option::is_none")] pub event_payload: Option, + /// Raw point-in-time event metadata for mark/system steps. + #[serde(skip_serializing_if = "Option::is_none")] + pub event_metadata: Option, + /// Semantic ATOF category for mark/system steps, when the mark carries one. + #[serde(skip_serializing_if = "Option::is_none")] + pub event_category: Option, + /// Vendor subtype for mark/system steps, from the mark category profile. + #[serde(skip_serializing_if = "Option::is_none")] + pub event_subtype: Option, /// Per-tool callable lineage, aligned with `tool_calls`. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub tool_ancestry: Vec, @@ -1888,6 +1897,9 @@ impl PendingAgentStep { llm_request: None, llm_response: self.llm_response.take(), event_payload: None, + event_metadata: None, + event_category: None, + event_subtype: None, tool_ancestry: std::mem::take(&mut self.tool_ancestry), tool_invocations: if self.tool_invocations.is_empty() { None @@ -2244,6 +2256,9 @@ impl StepConversionState { llm_request: Some(content.clone()), llm_response: None, event_payload: None, + event_metadata: None, + event_category: None, + event_subtype: None, tool_ancestry: Vec::new(), tool_invocations: None, }; @@ -2592,10 +2607,16 @@ impl StepConversionState { return; } self.flush_observations(); - let Some(data) = mark.data() else { - return; - }; - if is_empty_mark_payload(data) { + let payload = mark + .data() + .filter(|data| !is_empty_mark_payload(data)) + .cloned(); + let metadata = mark.metadata(); + let category = mark.category(); + // Preserve any mark that carries inspectable content: a payload, + // metadata, or a semantic category. Marks with none of these are runtime + // noise (for example empty checkpoints) and stay skipped. + if payload.is_none() && metadata.is_none() && category.is_none() { return; } let extra = AtifStepExtra { @@ -2609,14 +2630,19 @@ impl StepConversionState { }), llm_request: None, llm_response: None, - event_payload: Some(data.clone()), + event_payload: payload, + event_metadata: metadata.cloned(), + event_category: category.map(|category| category.as_str().to_string()), + event_subtype: mark + .category_profile() + .and_then(|profile| profile.subtype.clone()), tool_ancestry: Vec::new(), tool_invocations: None, }; self.steps.push(AtifStep { step_id: 0, source: "system".to_string(), - message: mark_message(mark, data), + message: mark_message(mark), timestamp: Some(mark.timestamp().to_rfc3339()), model_name: None, reasoning_effort: None, @@ -3352,7 +3378,7 @@ fn events_to_steps_for_agent( /// step with multiple results /// 4. Tool End observation results are correlated with the preceding LLM End's /// promoted tool_calls by function name → `source_call_id`. -/// 5. Mark events → system steps if they carry data. +/// 5. Mark events → system steps when they carry a payload, metadata, or category. /// 6. Scope Start/End → skipped. fn events_to_steps(events: &[&Event]) -> Vec { let mut sorted: Vec<&Event> = events.to_vec(); @@ -3384,8 +3410,9 @@ fn is_llm_chunk_mark(mark: &Event) -> bool { // A runtime mark is point-in-time telemetry rather than a scoped call with start/end events. Agent // hook adapters use marks for lifecycle notifications that do not map to first-class ATIF step // types, for example hook-only status updates or synthetic fallback events. The ATIF step message -// stays schema-compatible while the original payload is preserved in `Step.extra.event_payload`. -fn mark_message(mark: &Event, _data: &Json) -> Json { +// stays schema-compatible while the original payload, metadata, category, and subtype are +// preserved in `Step.extra` (`event_payload`, `event_metadata`, `event_category`, `event_subtype`). +fn mark_message(mark: &Event) -> Json { Json::String(mark_hook_event_name(mark).unwrap_or_default()) } diff --git a/crates/core/src/observability/openinference.rs b/crates/core/src/observability/openinference.rs index 649460c96..3fe51afc9 100644 --- a/crates/core/src/observability/openinference.rs +++ b/crates/core/src/observability/openinference.rs @@ -1108,6 +1108,21 @@ fn mark_attributes(event: &Event) -> Vec { "nemo_relay.mark.metadata_json", event.metadata(), ); + if let Some(category) = event.category() { + attributes.push(KeyValue::new( + "nemo_relay.mark.category", + category.as_str().to_string(), + )); + } + if let Some(subtype) = event + .category_profile() + .and_then(|profile| profile.subtype.as_deref()) + { + attributes.push(KeyValue::new( + "nemo_relay.mark.subtype", + subtype.to_string(), + )); + } attributes } diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 474af253b..1ce908684 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -645,6 +645,21 @@ fn mark_attributes(event: &Event) -> Vec { "nemo_relay.mark.metadata_json", event.metadata(), ); + if let Some(category) = event.category() { + attributes.push(KeyValue::new( + "nemo_relay.mark.category", + category.as_str().to_string(), + )); + } + if let Some(subtype) = event + .category_profile() + .and_then(|profile| profile.subtype.as_deref()) + { + attributes.push(KeyValue::new( + "nemo_relay.mark.subtype", + subtype.to_string(), + )); + } attributes } diff --git a/crates/core/tests/unit/atif_tests.rs b/crates/core/tests/unit/atif_tests.rs index 78e3c3b89..5ad394177 100644 --- a/crates/core/tests/unit/atif_tests.rs +++ b/crates/core/tests/unit/atif_tests.rs @@ -2670,6 +2670,45 @@ fn test_exporter_mark_steps_include_hook_name_and_ancestry() { ); } +#[test] +fn test_exporter_preserves_metadata_only_mark() { + let exporter = AtifExporter::new("session-1".to_string(), make_agent_info()); + let agent_uuid = Uuid::now_v7(); + let mark_uuid = Uuid::now_v7(); + + let agent_start = event_builder(agent_uuid, EventType::Start) + .name("agent") + .scope_type(ScopeType::Agent) + .build(); + // A payload-less mark that still carries metadata must be preserved rather + // than dropped as empty noise. + let mark = event_builder(mark_uuid, EventType::Mark) + .name("status-note") + .parent_uuid(agent_uuid) + .metadata(json!({"hook_event_name": "Notification", "detail": "waiting"})) + .build(); + + { + let mut state = exporter.state.lock().unwrap(); + state.events.push(agent_start); + state.events.push(mark); + } + + let trajectory = exporter.export().unwrap(); + let step = trajectory + .steps + .iter() + .find(|step| step.source == "system") + .expect("metadata-only mark should produce a system step"); + assert_eq!(step.message, json!("Notification")); + let extra: AtifStepExtra = serde_json::from_value(step.extra.clone().unwrap()).unwrap(); + assert!(extra.event_payload.is_none()); + assert_eq!( + extra.event_metadata, + Some(json!({"hook_event_name": "Notification", "detail": "waiting"})) + ); +} + #[test] fn test_exporter_embeds_nested_subagent_trajectory() { let root_uuid = Uuid::now_v7(); diff --git a/crates/core/tests/unit/observability/exporter_parity_tests.rs b/crates/core/tests/unit/observability/exporter_parity_tests.rs index 16cbbaf3c..0b79d77d6 100644 --- a/crates/core/tests/unit/observability/exporter_parity_tests.rs +++ b/crates/core/tests/unit/observability/exporter_parity_tests.rs @@ -11,14 +11,15 @@ use std::collections::HashMap; +use chrono::{DateTime, Duration, Utc}; use opentelemetry::KeyValue; use opentelemetry_sdk::trace::{InMemorySpanExporterBuilder, SdkTracerProvider, SpanData}; use serde_json::json; use uuid::Uuid; use crate::api::event::{ - BaseEvent, CategoryProfile, Event, EventCategory, EventNormalizationExt, ScopeCategory, - ScopeEvent, + BaseEvent, CategoryProfile, Event, EventCategory, EventNormalizationExt, MarkEvent, + ScopeCategory, ScopeEvent, }; use crate::codec::model_pricing::pricing_test_mutex; use crate::codec::response::{ @@ -791,3 +792,362 @@ fn test_manual_fallback_payload_parity() { Some(&"1500".to_string()) ); } + +// =================================================================== +// Mark projection parity +// =================================================================== + +fn parity_base_ts() -> DateTime { + DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z") + .unwrap() + .with_timezone(&Utc) +} + +fn scope_event_at( + scope_category: ScopeCategory, + uuid: Uuid, + name: &str, + category: EventCategory, + timestamp: DateTime, + data: Option, +) -> Event { + Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .uuid(uuid) + .name(name) + .timestamp(timestamp) + .data_opt(data) + .build(), + scope_category, + Vec::new(), + category, + None, + )) +} + +#[allow(clippy::too_many_arguments)] +fn mark_event_at( + uuid: Uuid, + parent: Option, + name: &str, + timestamp: DateTime, + data: Option, + metadata: Option, + category: Option, + profile: Option, +) -> Event { + Event::Mark(MarkEvent::new( + BaseEvent::builder() + .uuid(uuid) + .name(name) + .timestamp(timestamp) + .parent_uuid_opt(parent) + .data_opt(data) + .metadata_opt(metadata) + .build(), + category, + profile, + )) +} + +/// Attributes of the named span event recorded on the named span. +fn span_event_attrs( + spans: &[SpanData], + span_name: &str, + event_name: &str, +) -> HashMap { + let span = spans + .iter() + .find(|span| span.name.as_ref() == span_name) + .unwrap_or_else(|| panic!("missing span {span_name}")); + let event = span + .events + .events + .iter() + .find(|event| event.name.as_ref() == event_name) + .unwrap_or_else(|| panic!("missing span event {event_name} on span {span_name}")); + attr_map(&event.attributes) +} + +fn system_steps(trajectory: &AtifTrajectory) -> Vec<&AtifStep> { + trajectory + .steps + .iter() + .filter(|step| step.source == "system") + .collect() +} + +/// A mark with structured data, metadata, category, and subtype under an open +/// parent scope must retain every field across ATOF, ATIF, OTLP, and +/// OpenInference. +#[test] +fn test_mark_field_retention_parity_across_exporters() { + let agent_uuid = Uuid::now_v7(); + let mark_uuid = Uuid::now_v7(); + let data = json!({"checkpoint": "phase-1", "count": 3}); + let metadata = json!({"emitter": "neutral-fixture"}); + + let events = [ + scope_event_at( + ScopeCategory::Start, + agent_uuid, + "agent-scope", + EventCategory::agent(), + parity_base_ts(), + None, + ), + mark_event_at( + mark_uuid, + Some(agent_uuid), + "phase-marker", + parity_base_ts() + Duration::seconds(1), + Some(data.clone()), + Some(metadata.clone()), + Some(EventCategory::custom()), + Some(CategoryProfile::builder().subtype("checkpoint").build()), + ), + scope_event_at( + ScopeCategory::End, + agent_uuid, + "agent-scope", + EventCategory::agent(), + parity_base_ts() + Duration::seconds(2), + None, + ), + ]; + + // ATOF: the source event retains every field verbatim. + let atof = events[1].to_json_value(); + assert_eq!(atof["kind"], json!("mark")); + assert_eq!(atof["name"], json!("phase-marker")); + assert_eq!(atof["data"], data); + assert_eq!(atof["metadata"], metadata); + assert_eq!(atof["category"], json!("custom")); + assert_eq!(atof["category_profile"]["subtype"], json!("checkpoint")); + assert_eq!(atof["parent_uuid"], json!(agent_uuid.to_string())); + + let exports = export_through_all_exporters(&events); + + // ATIF: a system step preserves name, payload, metadata, category, subtype, + // parentage, and timestamp. + let steps = system_steps(&exports.trajectory); + assert_eq!(steps.len(), 1); + let step = steps[0]; + assert_eq!(step.message, json!("phase-marker")); + assert_eq!( + step.timestamp.as_deref(), + Some( + (parity_base_ts() + Duration::seconds(1)) + .to_rfc3339() + .as_str() + ) + ); + let extra: AtifStepExtra = serde_json::from_value(step.extra.clone().unwrap()).unwrap(); + assert_eq!(extra.event_payload, Some(data.clone())); + assert_eq!(extra.event_metadata, Some(metadata.clone())); + assert_eq!(extra.event_category.as_deref(), Some("custom")); + assert_eq!(extra.event_subtype.as_deref(), Some("checkpoint")); + assert_eq!(extra.ancestry.function_name, "phase-marker"); + assert_eq!( + extra.ancestry.parent_id.as_deref(), + Some(agent_uuid.to_string().as_str()) + ); + + // OTLP + OpenInference: the mark attaches to its open parent span as a span + // event that keeps the same facts, including category and subtype. + for attrs in [ + span_event_attrs(&exports.otel_spans, "agent-scope", "phase-marker"), + span_event_attrs(&exports.openinference_spans, "agent-scope", "phase-marker"), + ] { + assert_eq!( + attrs.get("nemo_relay.mark.uuid"), + Some(&mark_uuid.to_string()) + ); + assert_eq!( + attrs.get("nemo_relay.mark.parent_uuid"), + Some(&agent_uuid.to_string()) + ); + assert_eq!( + attrs + .get("nemo_relay.mark.data_json") + .map(|raw| serde_json::from_str::(raw).unwrap()), + Some(data.clone()) + ); + assert_eq!( + attrs + .get("nemo_relay.mark.metadata_json") + .map(|raw| serde_json::from_str::(raw).unwrap()), + Some(metadata.clone()) + ); + assert_eq!( + attrs.get("nemo_relay.mark.category"), + Some(&"custom".to_string()) + ); + assert_eq!( + attrs.get("nemo_relay.mark.subtype"), + Some(&"checkpoint".to_string()) + ); + } +} + +/// A mark emitted after an LLM call keeps its position relative to the LLM +/// steps in the ATIF trajectory. +#[test] +fn test_mark_ordering_preserved_relative_to_llm_in_atif() { + let agent_uuid = Uuid::now_v7(); + let llm_uuid = Uuid::now_v7(); + let mark_uuid = Uuid::now_v7(); + + let events = [ + scope_event_at( + ScopeCategory::Start, + agent_uuid, + "agent-scope", + EventCategory::agent(), + parity_base_ts(), + None, + ), + scope_event_at( + ScopeCategory::Start, + llm_uuid, + "model-call", + EventCategory::llm(), + parity_base_ts() + Duration::seconds(1), + Some(json!({"headers": {}, "content": chat_request_content("order-model")})), + ), + scope_event_at( + ScopeCategory::End, + llm_uuid, + "model-call", + EventCategory::llm(), + parity_base_ts() + Duration::seconds(2), + Some(chat_response_output("order-model")), + ), + mark_event_at( + mark_uuid, + Some(agent_uuid), + "post-call-note", + parity_base_ts() + Duration::seconds(3), + Some(json!({"note": "after the model call"})), + None, + None, + None, + ), + scope_event_at( + ScopeCategory::End, + agent_uuid, + "agent-scope", + EventCategory::agent(), + parity_base_ts() + Duration::seconds(4), + None, + ), + ]; + + let exports = export_through_all_exporters(&events); + let sources: Vec<&str> = exports + .trajectory + .steps + .iter() + .map(|step| step.source.as_str()) + .collect(); + // LLM start -> user, LLM end -> agent, mark -> system, in emission order. + assert_eq!(sources, ["user", "agent", "system"]); + assert_eq!( + system_steps(&exports.trajectory)[0].message, + json!("post-call-note") + ); +} + +/// Marks emitted after their parent scope has closed (or with no parent) still +/// reach every exporter: ATIF keeps them as discoverable system steps and the +/// span exporters fall back to zero-duration orphan spans that retain the mark +/// facts. +#[test] +fn test_orphan_mark_fallback_parity() { + let agent_uuid = Uuid::now_v7(); + let late_uuid = Uuid::now_v7(); + let detached_uuid = Uuid::now_v7(); + + let events = [ + scope_event_at( + ScopeCategory::Start, + agent_uuid, + "agent-scope", + EventCategory::agent(), + parity_base_ts(), + None, + ), + scope_event_at( + ScopeCategory::End, + agent_uuid, + "agent-scope", + EventCategory::agent(), + parity_base_ts() + Duration::seconds(1), + None, + ), + // Parent scope already closed when this mark is emitted. + mark_event_at( + late_uuid, + Some(agent_uuid), + "late-note", + parity_base_ts() + Duration::seconds(2), + Some(json!({"stage": "post-scope"})), + None, + Some(EventCategory::custom()), + Some(CategoryProfile::builder().subtype("late").build()), + ), + // No parent scope at all. + mark_event_at( + detached_uuid, + None, + "detached-note", + parity_base_ts() + Duration::seconds(3), + Some(json!({"stage": "standalone"})), + None, + None, + None, + ), + ]; + + let exports = export_through_all_exporters(&events); + + // ATIF keeps the after-scope mark as a system step with its facts intact. + let late_step = system_steps(&exports.trajectory) + .into_iter() + .find(|step| step.message == json!("late-note")) + .expect("ATIF should retain the after-scope mark as a system step"); + let late_extra: AtifStepExtra = + serde_json::from_value(late_step.extra.clone().unwrap()).unwrap(); + assert_eq!(late_extra.event_category.as_deref(), Some("custom")); + assert_eq!(late_extra.event_subtype.as_deref(), Some("late")); + + // Both span exporters fall back to zero-duration orphan spans that retain + // the mark facts, including category/subtype. + for spans in [&exports.otel_spans, &exports.openinference_spans] { + let late = spans + .iter() + .find(|span| span.name.as_ref() == "mark:late-note") + .expect("orphan span mark:late-note"); + assert_eq!(late.start_time, late.end_time); + let attrs = attr_map(&late.attributes); + assert_eq!( + attrs.get("nemo_relay.mark.orphan"), + Some(&"true".to_string()) + ); + assert_eq!( + attrs.get("nemo_relay.mark.category"), + Some(&"custom".to_string()) + ); + assert_eq!( + attrs.get("nemo_relay.mark.subtype"), + Some(&"late".to_string()) + ); + + assert!( + spans + .iter() + .any(|span| span.name.as_ref() == "mark:detached-note"), + "parentless mark should still produce an orphan span" + ); + } +} diff --git a/docs/observability-plugin/atif.mdx b/docs/observability-plugin/atif.mdx index 847d8c9c6..5f6940bf8 100644 --- a/docs/observability-plugin/atif.mdx +++ b/docs/observability-plugin/atif.mdx @@ -183,6 +183,12 @@ per-destination sink failures are isolated to the failed sink. The exporter translates NeMo Relay lifecycle events into ATIF v1.7 trajectory data. LLM start and end events become model steps, tool events become tool calls and observations, and scope nesting contributes lineage metadata. +Point-in-time mark events become `system` steps: the step `message` is the +mark's hook event name when present, otherwise the mark name, and the mark +payload, metadata, semantic category, and subtype are preserved under +`extra.event_payload`, `extra.event_metadata`, `extra.event_category`, and +`extra.event_subtype`. A mark is recorded whenever it carries a payload, +metadata, or a category, so a mark stays discoverable even without a payload. Nested agent scopes are embedded in the parent file as `subagent_trajectories` and referenced from parent observation results with `subagent_trajectory_ref.trajectory_id`. The reference points to the embedded