feat: add cross-boundary Relay propagation

crates/core/src/api/llm.rs

@@ -22,6 +22,7 @@ use crate::api::runtime::global_context;
 use crate::api::runtime::{
     EventSubscriberFn, LlmCollectorFn, LlmExecutionNextFn, LlmFinalizerFn, LlmJsonStream,
     LlmSanitizeRequestContext, LlmSanitizeResponseContext, LlmStreamExecutionNextFn,
+    with_active_event_uuid,
 };
 use crate::api::runtime::{ScopeStackHandle, current_scope_stack};
 use crate::api::scope::event;
@@ -1025,7 +1026,9 @@ pub async fn llm_call_execute(params: LlmCallExecuteParams) -> Result<Json> {
     emit_optimization_marks(&handle, &lifecycle_subscribers);
 
     let execution_name = name.clone();
-    let execution =
+    let event_uuid = handle.uuid;
+    let execution = with_active_event_uuid(
+        event_uuid,
         scope_llm_optimization_recorder(handle.optimization_recorder.clone(), async move {
             let execution = {
                 let scope_stack = current_scope_stack();
@@ -1040,8 +1043,9 @@ pub async fn llm_call_execute(params: LlmCallExecuteParams) -> Result<Json> {
                 state.llm_build_execution_chain(&execution_name, func, &scope_locals)
             };
             execution(intercepted_request).await
-        })
-        .await;
+        }),
+    )
+    .await;
 
     match execution {
         Ok(response) => {
@@ -1215,7 +1219,9 @@ pub async fn llm_stream_call_execute(params: LlmStreamCallExecuteParams) -> Resu
     emit_optimization_marks(&handle, &lifecycle_subscribers);
 
     let execution_name = name.clone();
-    let execution =
+    let event_uuid = handle.uuid;
+    let execution = with_active_event_uuid(
+        event_uuid,
         scope_llm_optimization_recorder(handle.optimization_recorder.clone(), async move {
             let execution = {
                 let scope_stack = current_scope_stack();
@@ -1230,8 +1236,9 @@ pub async fn llm_stream_call_execute(params: LlmStreamCallExecuteParams) -> Resu
                 state.llm_stream_build_execution_chain(&execution_name, func, &scope_locals)
             };
             execution(intercepted_request).await
-        })
-        .await;
+        }),
+    )
+    .await;
 
     match execution {
         Ok(raw_stream) => {

crates/core/src/api/runtime.rs

@@ -19,10 +19,12 @@ pub use callbacks::{
 };
 pub use global::global_context;
 pub use scope_stack::{
-    ScopeStack, ScopeStackHandle, TASK_SCOPE_STACK, ThreadScopeStackBinding,
-    capture_thread_scope_stack, create_scope_stack, current_scope_stack, propagate_scope_to_thread,
-    restore_thread_scope_stack, scope_stack_active, set_thread_scope_stack,
-    sync_thread_scope_stack, task_scope_push, task_scope_remove, task_scope_top, with_scope_stack,
+    PropagationContext, ScopeStack, ScopeStackHandle, TASK_SCOPE_STACK, ThreadScopeStackBinding,
+    capture_propagation_context, capture_propagation_context_with_root, capture_thread_scope_stack,
+    create_scope_stack, create_scope_stack_from_propagation, current_scope_stack,
+    propagate_scope_to_thread, restore_thread_scope_stack, scope_stack_active,
+    set_thread_scope_stack, sync_thread_scope_stack, task_scope_push, task_scope_remove,
+    task_scope_top, with_active_event_uuid, with_scope_stack,
 };
 pub use state::NemoRelayContextState;
 pub use subscriber_dispatcher::flush_subscribers;

crates/core/src/api/runtime/scope_stack.rs

@@ -10,8 +10,10 @@
 
 use std::cell::RefCell;
 use std::collections::{HashMap, HashSet};
+use std::future::Future;
 use std::sync::{Arc, RwLock};
 
+use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
 use crate::api::runtime::callbacks::EventSubscriberFn;
@@ -31,6 +33,66 @@ pub struct ScopeStack {
     stack: Vec<ScopeHandle>,
     scope_registries: HashMap<Uuid, ScopeLocalRegistries>,
     fresh_agents: HashSet<Uuid>,
+    propagated_parent_uuid: Option<Uuid>,
+}
+
+/// Versioned, transport-neutral causal context for crossing a Relay boundary.
+///
+/// Applications are responsible for serializing, transporting, authenticating,
+/// and trusting this value. It intentionally contains only Relay identifiers;
+/// OpenTelemetry `traceparent` and `tracestate` remain transport sidecars.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PropagationContext {
+    /// Wire-format version. Version 1 is the only currently supported value.
+    pub version: u16,
+    /// Stable session root when the sending application knows one.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub root_uuid: Option<Uuid>,
+    /// Immediate Relay event or scope that caused the boundary crossing.
+    pub parent_uuid: Uuid,
+}
+
+impl PropagationContext {
+    /// The current wire-format version.
+    pub const VERSION: u16 = 1;
+
+    /// Serialize this validated context for application-managed transport.
+    pub fn to_json(&self) -> Result<String> {
+        self.validate()?;
+        Ok(serde_json::to_string(self).expect("PropagationContext is always JSON serializable"))
+    }
+
+    /// Deserialize and validate a context received from application-managed transport.
+    pub fn from_json(value: &str) -> Result<Self> {
+        let context: Self = serde_json::from_str(value).map_err(|error| {
+            FlowError::InvalidArgument(format!("invalid propagation context JSON: {error}"))
+        })?;
+        context.validate()?;
+        Ok(context)
+    }
+
+    /// Validate a context received from an untrusted transport.
+    pub fn validate(&self) -> Result<()> {
+        if self.version != Self::VERSION {
+            return Err(FlowError::InvalidArgument(format!(
+                "unsupported propagation context version {}; expected {}",
+                self.version,
+                Self::VERSION
+            )));
+        }
+        for (name, uuid) in [("parent_uuid", self.parent_uuid)]
+            .into_iter()
+            .chain(self.root_uuid.map(|uuid| ("root_uuid", uuid)))
+        {
+            let bytes = uuid.as_bytes();
+            if bytes.iter().all(|byte| *byte == 0) || bytes[8..].iter().all(|byte| *byte == 0) {
+                return Err(FlowError::InvalidArgument(format!(
+                    "propagation context {name} is not a usable Relay identifier"
+                )));
+            }
+        }
+        Ok(())
+    }
 }
 
 impl ScopeStack {
@@ -49,7 +111,49 @@ impl ScopeStack {
             stack: vec![root],
             scope_registries: HashMap::new(),
             fresh_agents: HashSet::from([root_uuid]),
+            propagated_parent_uuid: None,
+        }
+    }
+
+    fn from_propagation(context: &PropagationContext) -> Result<Self> {
+        context.validate()?;
+        let (root, parent) = match context.root_uuid {
+            Some(root_uuid) => {
+                let root = ScopeHandle::builder()
+                    .uuid(root_uuid)
+                    .name("propagated-root")
+                    .scope_type(ScopeType::Agent)
+                    .build();
+                let parent = (root_uuid != context.parent_uuid).then(|| {
+                    ScopeHandle::builder()
+                        .uuid(context.parent_uuid)
+                        .parent_uuid(root_uuid)
+                        .name("propagated-parent")
+                        .scope_type(ScopeType::Unknown)
+                        .build()
+                });
+                (root, parent)
+            }
+            None => (
+                ScopeHandle::builder()
+                    .uuid(context.parent_uuid)
+                    .name("propagated-root")
+                    .scope_type(ScopeType::Agent)
+                    .build(),
+                None,
+            ),
+        };
+        let root_uuid = root.uuid;
+        let mut stack = vec![root];
+        if let Some(parent) = parent {
+            stack.push(parent);
         }
+        Ok(Self {
+            stack,
+            scope_registries: HashMap::new(),
+            fresh_agents: HashSet::from([root_uuid]),
+            propagated_parent_uuid: context.root_uuid.map(|_| context.parent_uuid),
+        })
     }
 
     /// Push a scope handle onto the top of the stack.
@@ -98,6 +202,11 @@ impl ScopeStack {
             .uuid
     }
 
+    /// Whether `uuid` is the synthetic parent imported from propagation.
+    pub fn is_propagated_parent(&self, uuid: Uuid) -> bool {
+        self.propagated_parent_uuid == Some(uuid)
+    }
+
     /// Return the full ordered stack of scope handles.
     ///
     /// # Returns
@@ -276,6 +385,13 @@ pub struct ThreadScopeStackBinding {
     explicit: bool,
 }
 
+impl ThreadScopeStackBinding {
+    /// Return the captured thread-local scope stack handle.
+    pub fn stack(&self) -> ScopeStackHandle {
+        self.stack.clone()
+    }
+}
+
 /// Create a new scope stack handle with an implicit root scope.
 ///
 /// The returned handle wraps a freshly initialized [`ScopeStack`] inside an
@@ -290,9 +406,48 @@ pub fn create_scope_stack() -> ScopeStackHandle {
     Arc::new(RwLock::new(ScopeStack::new()))
 }
 
+/// Create an isolated scope stack rooted below a supplied propagation context.
+///
+/// The imported handles are synthetic bookkeeping only; Relay never emits their
+/// lifecycle events or transfers scope-local registrations across the boundary.
+pub fn create_scope_stack_from_propagation(
+    context: &PropagationContext,
+) -> Result<ScopeStackHandle> {
+    Ok(Arc::new(RwLock::new(ScopeStack::from_propagation(
+        context,
+    )?)))
+}
+
+/// Capture the current causal parent without asserting a session root.
+pub fn capture_propagation_context() -> Result<PropagationContext> {
+    capture_propagation_context_with_root(None)
+}
+
+/// Capture the current causal parent and an application-supplied session root.
+pub fn capture_propagation_context_with_root(
+    root_uuid: Option<Uuid>,
+) -> Result<PropagationContext> {
+    let context = PropagationContext {
+        version: PropagationContext::VERSION,
+        root_uuid,
+        parent_uuid: ACTIVE_EVENT_UUID
+            .try_with(|uuid| *uuid)
+            .unwrap_or_else(|_| task_scope_top().uuid),
+    };
+    context.validate()?;
+    Ok(context)
+}
+
 tokio::task_local! {
     /// Task-local scope stack handle used by async execution contexts.
     pub static TASK_SCOPE_STACK: ScopeStackHandle;
+    /// Managed tool or LLM event currently executing in this task.
+    static ACTIVE_EVENT_UUID: Uuid;
+}
+
+/// Run a future with `uuid` as the causally active managed event.
+pub async fn with_active_event_uuid<T>(uuid: Uuid, future: impl Future<Output = T>) -> T {
+    ACTIVE_EVENT_UUID.scope(uuid, future).await
 }
 
 thread_local! {

crates/core/src/api/tool.rs

@@ -7,7 +7,7 @@ use crate::api::event::{BaseEvent, Event, MarkEvent, PendingMarkSpec};
 use crate::api::runtime::NemoRelayContextState;
 use crate::api::runtime::current_scope_stack;
 use crate::api::runtime::global_context;
-use crate::api::runtime::{EventSubscriberFn, ToolExecutionNextFn};
+use crate::api::runtime::{EventSubscriberFn, ToolExecutionNextFn, with_active_event_uuid};
 use crate::api::scope::event;
 use crate::api::scope::{EmitMarkEventParams, ScopeHandle};
 use crate::api::shared::{
@@ -551,7 +551,7 @@ pub async fn tool_call_execute(params: ToolCallExecuteParams) -> Result<Json> {
         state.tool_build_execution_chain(&name, func, &scope_locals)
     };
 
-    match execution(intercepted_args).await {
+    match with_active_event_uuid(handle.uuid, execution(intercepted_args)).await {
         Ok(outcome) => {
             let ToolExecutionInterceptOutcome {
                 result,

crates/core/src/observability/openinference.rs

@@ -29,7 +29,7 @@ use super::{
     relay_trace_id, validate_attribute_mappings,
 };
 use crate::api::event::{Event, EventNormalizationExt, ScopeCategory};
-use crate::api::runtime::EventSubscriberFn;
+use crate::api::runtime::{EventSubscriberFn, current_scope_stack};
 use crate::api::scope::ScopeType;
 use crate::api::subscriber::{deregister_subscriber, flush_subscribers, register_subscriber};
 use crate::codec::request::{
@@ -42,7 +42,8 @@ use chrono::{DateTime, Utc};
 use openinference_semantic_conventions::SpanKind as OpenInferenceSpanKind;
 use openinference_semantic_conventions::attributes as oi;
 use opentelemetry::trace::{
-    Span as _, SpanContext, SpanKind, TraceContextExt, Tracer, TracerProvider as _,
+    Span as _, SpanContext, SpanKind, TraceContextExt, TraceFlags, TraceState, Tracer,
+    TracerProvider as _,
 };
 use opentelemetry::{Context, KeyValue};
 use opentelemetry_otlp::{Protocol, SpanExporter, WithExportConfig, WithHttpConfig};
@@ -788,11 +789,28 @@ impl OpenInferenceEventProcessor {
         if let Some(active_span) = self.find_parent_span(event) {
             return Context::new().with_remote_span_context(active_span.span_context.clone());
         }
-        event
+        if let Some(span_context) = event
             .parent_uuid()
             .and_then(|uuid| self.completed_span_contexts.get(&uuid))
-            .map(|span_context| Context::new().with_remote_span_context(span_context.clone()))
-            .unwrap_or_default()
+        {
+            return Context::new().with_remote_span_context(span_context.clone());
+        }
+        let Some(parent_uuid) = event.parent_uuid() else {
+            return Context::new();
+        };
+        let stack = current_scope_stack();
+        let stack = stack.read().expect("scope stack lock poisoned");
+        if !stack.is_propagated_parent(parent_uuid) {
+            return Context::new();
+        }
+        let root_uuid = stack.root_uuid();
+        Context::new().with_remote_span_context(SpanContext::new(
+            relay_trace_id(root_uuid),
+            relay_span_id(parent_uuid),
+            TraceFlags::SAMPLED,
+            true,
+            TraceState::default(),
+        ))
     }
 
     fn parent_span_uuid(&self, event: &Event) -> Option<Uuid> {

crates/core/src/observability/otel.rs

@@ -29,14 +29,15 @@ use super::{
     relay_trace_id, validate_attribute_mappings,
 };
 use crate::api::event::{Event, EventNormalizationExt, ScopeCategory};
-use crate::api::runtime::EventSubscriberFn;
+use crate::api::runtime::{EventSubscriberFn, current_scope_stack};
 use crate::api::scope::ScopeType;
 use crate::api::subscriber::{deregister_subscriber, flush_subscribers, register_subscriber};
 use crate::codec::response::CostEstimate;
 use crate::error::FlowError;
 use chrono::{DateTime, Utc};
 use opentelemetry::trace::{
-    Span as _, SpanContext, SpanKind, TraceContextExt, Tracer, TracerProvider as _,
+    Span as _, SpanContext, SpanKind, TraceContextExt, TraceFlags, TraceState, Tracer,
+    TracerProvider as _,
 };
 use opentelemetry::{Context, KeyValue};
 use opentelemetry_otlp::{Protocol, SpanExporter, WithExportConfig, WithHttpConfig};
@@ -774,11 +775,28 @@ impl OtelEventProcessor {
         if let Some(active_span) = self.find_parent_span(event) {
             return Context::new().with_remote_span_context(active_span.span_context.clone());
         }
-        event
+        if let Some(span_context) = event
             .parent_uuid()
             .and_then(|uuid| self.completed_span_contexts.get(&uuid))
-            .map(|span_context| Context::new().with_remote_span_context(span_context.clone()))
-            .unwrap_or_default()
+        {
+            return Context::new().with_remote_span_context(span_context.clone());
+        }
+        let Some(parent_uuid) = event.parent_uuid() else {
+            return Context::new();
+        };
+        let stack = current_scope_stack();
+        let stack = stack.read().expect("scope stack lock poisoned");
+        if !stack.is_propagated_parent(parent_uuid) {
+            return Context::new();
+        }
+        let root_uuid = stack.root_uuid();
+        Context::new().with_remote_span_context(SpanContext::new(
+            relay_trace_id(root_uuid),
+            relay_span_id(parent_uuid),
+            TraceFlags::SAMPLED,
+            true,
+            TraceState::default(),
+        ))
     }
 
     fn parent_span_uuid(&self, event: &Event) -> Option<Uuid> {