diff --git a/.changeset/defined-native-trace-flags.md b/.changeset/defined-native-trace-flags.md new file mode 100644 index 000000000..14a297d5e --- /dev/null +++ b/.changeset/defined-native-trace-flags.md @@ -0,0 +1,6 @@ +--- +'@openai/agents': patch +'@openai/agents-codex': patch +--- + +fix: accept defined sampled and random-trace-ID flag combinations and preserve them through native parent and callback header propagation, including later owned child turns. diff --git a/packages/agents-codex/native/patches/0058-w3c-defined-trace-flags.patch b/packages/agents-codex/native/patches/0058-w3c-defined-trace-flags.patch new file mode 100644 index 000000000..9768d35c0 --- /dev/null +++ b/packages/agents-codex/native/patches/0058-w3c-defined-trace-flags.patch @@ -0,0 +1,294 @@ +diff --git a/codex-rs/otel/src/trace_context.rs b/codex-rs/otel/src/trace_context.rs +index e1c62777f1c1bd33eaf6898a57b9157de0085e79..bc5c1f256f8e5419c64508b25383fb2300e6a2ae 100644 +--- a/codex-rs/otel/src/trace_context.rs ++++ b/codex-rs/otel/src/trace_context.rs +@@ -9,7 +9,10 @@ use std::sync::RwLock; + use codex_protocol::protocol::W3cTraceContext; + use opentelemetry::Context; + use opentelemetry::propagation::TextMapPropagator; ++use opentelemetry::trace::SpanContext; + use opentelemetry::trace::TraceContextExt; ++use opentelemetry::trace::TraceFlags; ++use opentelemetry::trace::TraceId; + use opentelemetry::trace::TraceState; + use opentelemetry_sdk::propagation::TraceContextPropagator; + use tracing::Span; +@@ -17,6 +20,13 @@ use tracing::debug; + use tracing::warn; + use tracing_opentelemetry::OpenTelemetrySpanExt; + ++// The pinned SDK clears flags on dropped spans. Bind the inherited random bit ++// to its trace identity so it cannot leak when a Context adopts another trace. ++struct RandomTraceId { ++ trace_id: TraceId, ++ random: bool, ++} ++ + const TRACEPARENT_ENV_VAR: &str = "TRACEPARENT"; + const TRACESTATE_ENV_VAR: &str = "TRACESTATE"; + static TRACEPARENT_CONTEXT: OnceLock> = OnceLock::new(); +@@ -42,6 +52,22 @@ pub(crate) fn context_w3c_trace_context(context: &Context) -> Option() ++ .filter(|random| random.trace_id == context.span().span_context().trace_id()) ++ { ++ flags = (flags & !0x02) | if random.random { 0x02 } else { 0 }; ++ } ++ if flags & 0x02 != 0 { ++ if let Some(parent) = headers.get_mut("traceparent") { ++ if let Some((prefix, _)) = parent.rsplit_once('-') { ++ *parent = format!("{prefix}-{flags:02x}"); ++ } ++ } ++ } + let tracestate = headers.remove("tracestate"); + let configured_tracestate_guard = tracestate_entries() + .read() +@@ -122,7 +148,23 @@ pub fn set_parent_from_w3c_trace_context(span: &Span, trace: &W3cTraceContext) - + } + + pub fn set_parent_from_context(span: &Span, context: Context) { +- let _ = span.set_parent(context); ++ let _ = span.set_parent(with_inherited_random_flag(context)); ++} ++ ++fn with_inherited_random_flag(context: Context) -> Context { ++ let parent = context.span().span_context().clone(); ++ let context = ++ if parent.is_valid() && (parent.is_remote() || parent.trace_flags().to_u8() & 0x02 != 0) { ++ // Remote adoption is authoritative, including an explicitly unset bit. ++ // Local dropped children retain the inherited value in their Context. ++ context.with_value(RandomTraceId { ++ trace_id: parent.trace_id(), ++ random: parent.trace_flags().to_u8() & 0x02 != 0, ++ }) ++ } else { ++ context ++ }; ++ context + } + + pub fn traceparent_context_from_env() -> Option { +@@ -135,9 +177,36 @@ pub(crate) fn context_from_trace_headers( + traceparent: Option<&str>, + tracestate: Option<&str>, + ) -> Option { +- let traceparent = traceparent?; ++ let traceparent = traceparent?.trim(); ++ // The pinned propagator accepts short hexadecimal IDs. Require the W3C ++ // field widths, then delegate all remaining validation to the propagator. ++ let mut fields = traceparent.split('-'); ++ let version = fields.next()?; ++ let trace_id = fields.next()?; ++ let parent_id = fields.next()?; ++ let raw_flags = fields.next()?; ++ if version.len() != 2 || trace_id.len() != 32 || parent_id.len() != 16 || raw_flags.len() != 2 { ++ return None; ++ } ++ // W3C Level 2 defines bit 0 as sampled and bit 1 as random trace ID. ++ // Let the pinned propagator validate all other header syntax using its ++ // supported sampling bit, then restore the complete defined flags. ++ let flags = match raw_flags { ++ "02" => Some(0x02), ++ "03" => Some(0x03), ++ _ => None, ++ }; ++ let normalized = flags.and_then(|flags| { ++ traceparent.strip_prefix("00-")?; ++ traceparent ++ .strip_suffix(if flags == 2 { "02" } else { "03" }) ++ .map(|prefix| format!("{prefix}{:02x}", flags & 1)) ++ }); + let mut headers = HashMap::new(); +- headers.insert("traceparent".to_string(), traceparent.to_string()); ++ headers.insert( ++ "traceparent".to_string(), ++ normalized.unwrap_or_else(|| traceparent.to_string()), ++ ); + if let Some(tracestate) = tracestate { + headers.insert("tracestate".to_string(), tracestate.to_string()); + } +@@ -147,7 +216,20 @@ pub(crate) fn context_from_trace_headers( + if !context.span().span_context().is_valid() { + return None; + } +- Some(context) ++ if let Some(flags) = flags { ++ let span = context.span(); ++ let parent = span.span_context(); ++ return Some(with_inherited_random_flag( ++ context.with_remote_span_context(SpanContext::new( ++ parent.trace_id(), ++ parent.span_id(), ++ TraceFlags::new(flags), ++ true, ++ parent.trace_state().clone(), ++ )), ++ )); ++ } ++ Some(with_inherited_random_flag(context)) + } + + fn load_traceparent_context() -> Option { +@@ -373,6 +455,155 @@ mod tests { + assert!(span_context.is_remote()); + } + ++ #[test] ++ fn defined_trace_flags_survive_child_header_propagation() { ++ use super::span_w3c_trace_context; ++ use tracing_opentelemetry::OpenTelemetrySpanExt; ++ ++ for flags in ["00", "01", "02", "03"] { ++ let provider = SdkTracerProvider::builder().build(); ++ let subscriber = tracing_subscriber::registry() ++ .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("flags"))); ++ tracing::subscriber::with_default(subscriber, || { ++ let parent = ++ format!("00-11111111111111111111111111111111-2222222222222222-{flags}"); ++ let context = ++ context_from_trace_headers(Some(&parent), Some("vendor=value")).unwrap(); ++ assert_eq!( ++ context.span().span_context().trace_flags().to_u8(), ++ u8::from_str_radix(flags, 16).unwrap() ++ ); ++ let span = tracing::info_span!("child"); ++ super::set_parent_from_context(&span, context); ++ let headers = span_w3c_trace_context(&span).unwrap(); ++ let header = headers.traceparent.unwrap(); ++ assert_eq!(&header[3..35], &parent[3..35]); ++ assert_eq!(&header[53..], flags); ++ assert_ne!(&header[36..52], &parent[36..52]); ++ let grandchild = tracing::info_span!("grandchild"); ++ super::set_parent_from_context(&grandchild, span.context()); ++ let grandchild_header = span_w3c_trace_context(&grandchild) ++ .unwrap() ++ .traceparent ++ .unwrap(); ++ assert_eq!(&grandchild_header[3..35], &parent[3..35]); ++ assert_eq!(&grandchild_header[53..], flags); ++ assert_ne!(&grandchild_header[36..52], &header[36..52]); ++ assert_eq!(headers.tracestate.as_deref(), Some("vendor=value")); ++ }); ++ } ++ assert!( ++ context_from_trace_headers( ++ Some("00-00000000000000000000000000000000-2222222222222222-03"), ++ None ++ ) ++ .is_none() ++ ); ++ assert!(context_from_trace_headers(Some("00-bad-2222222222222222-03"), None).is_none()); ++ } ++ ++ #[test] ++ fn inherited_random_flag_preserves_local_sampling_decisions() { ++ for (sampler, incoming, outgoing) in [ ++ (opentelemetry_sdk::trace::Sampler::AlwaysOff, "03", "02"), ++ (opentelemetry_sdk::trace::Sampler::AlwaysOn, "02", "03"), ++ ] { ++ let provider = SdkTracerProvider::builder().with_sampler(sampler).build(); ++ let subscriber = tracing_subscriber::registry() ++ .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("sampling"))); ++ tracing::subscriber::with_default(subscriber, || { ++ let parent = ++ format!("00-11111111111111111111111111111111-2222222222222222-{incoming}"); ++ let context = context_from_trace_headers(Some(&parent), None).unwrap(); ++ let span = tracing::info_span!("sampling"); ++ super::set_parent_from_context(&span, context); ++ let header = super::span_w3c_trace_context(&span) ++ .unwrap() ++ .traceparent ++ .unwrap(); ++ assert_eq!(&header[53..], outgoing); ++ assert_eq!(&header[3..35], &parent[3..35]); ++ assert_ne!(&header[36..52], &parent[36..52]); ++ }); ++ } ++ } ++ ++ #[test] ++ fn inherited_random_flag_does_not_cross_trace_ids_or_fresh_roots() { ++ use tracing_opentelemetry::OpenTelemetrySpanExt; ++ let provider = SdkTracerProvider::builder().build(); ++ let subscriber = tracing_subscriber::registry() ++ .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("identity"))); ++ tracing::subscriber::with_default(subscriber, || { ++ let parent = "00-11111111111111111111111111111111-2222222222222222-02"; ++ let span = tracing::info_span!("original"); ++ super::set_parent_from_context( ++ &span, ++ context_from_trace_headers(Some(parent), None).unwrap(), ++ ); ++ for (trace_id, flags) in [ ++ ("11111111111111111111111111111111", 0), ++ ("11111111111111111111111111111111", 1), ++ ("33333333333333333333333333333333", 1), ++ ] { ++ let replacement = span.context().with_remote_span_context( ++ opentelemetry::trace::SpanContext::new( ++ TraceId::from_hex(trace_id).unwrap(), ++ SpanId::from_hex("4444444444444444").unwrap(), ++ opentelemetry::trace::TraceFlags::new(flags), ++ true, ++ Default::default(), ++ ), ++ ); ++ let other = tracing::info_span!("other"); ++ super::set_parent_from_context(&other, replacement); ++ let header = super::span_w3c_trace_context(&other) ++ .unwrap() ++ .traceparent ++ .unwrap(); ++ assert_eq!(&header[3..35], trace_id); ++ assert_eq!(&header[53..], format!("{flags:02x}")); ++ assert_ne!(&header[36..52], "4444444444444444"); ++ } ++ let invalid = tracing::info_span!("invalid"); ++ super::set_parent_from_context( ++ &invalid, ++ span.context() ++ .with_remote_span_context(opentelemetry::trace::SpanContext::empty_context()), ++ ); ++ assert!(super::span_w3c_trace_context(&invalid).is_none()); ++ let fresh = tracing::info_span!("fresh"); ++ super::set_parent_from_context(&fresh, opentelemetry::Context::new()); ++ let header = super::span_w3c_trace_context(&fresh) ++ .unwrap() ++ .traceparent ++ .unwrap(); ++ assert_ne!(&header[3..35], &parent[3..35]); ++ assert_eq!(&header[53..], "01"); ++ }); ++ } ++ ++ #[test] ++ fn direct_native_context_keeps_unsampled_random_trace_flag() { ++ use opentelemetry::trace::Tracer as _; ++ let provider = SdkTracerProvider::builder().build(); ++ let tracer = provider.tracer("direct"); ++ let parent = context_from_trace_headers( ++ Some("00-11111111111111111111111111111111-2222222222222222-02"), ++ None, ++ ) ++ .unwrap(); ++ let child = tracer.start_with_context("direct-child", &parent); ++ let context = parent.with_span(child); ++ let header = super::context_w3c_trace_context(&context) ++ .unwrap() ++ .traceparent ++ .unwrap(); ++ assert_eq!(&header[3..35], "11111111111111111111111111111111"); ++ assert_ne!(&header[36..52], "2222222222222222"); ++ assert_eq!(&header[53..], "02"); ++ } ++ + #[test] + fn invalid_traceparent_returns_none() { + let parent = context_from_trace_headers( diff --git a/packages/agents-codex/native/patches/series.json b/packages/agents-codex/native/patches/series.json index 8beb7cfb6..8532f81fa 100644 --- a/packages/agents-codex/native/patches/series.json +++ b/packages/agents-codex/native/patches/series.json @@ -271,7 +271,11 @@ { "path": "0059-native-request-metadata.patch", "sha256": "00fb1ca9eff806e34ebd56521d4367c4905e6be285b773a2c32cc8ec617f17ae" + }, + { + "path": "0058-w3c-defined-trace-flags.patch", + "sha256": "1c689d4839940e41ceb5f47c3c65d02a51608a02c63230ed435997a1c751b6ed" } ], - "result_tree": "6e3ca2e842c1caf783abd91fb2aed74233b8366d" + "result_tree": "a58aca9912ec58771e46f1fd577395486798cfbf" } diff --git a/packages/agents-codex/native/rust/lib.rs b/packages/agents-codex/native/rust/lib.rs index 22df3fe10..5136452d1 100644 --- a/packages/agents-codex/native/rust/lib.rs +++ b/packages/agents-codex/native/rust/lib.rs @@ -3950,44 +3950,46 @@ mod native_trace_merge_tests { ) .unwrap(); let token = readback["handle"].as_str().unwrap(); - let explicit = W3cTraceContext { - traceparent: Some(format!( - "00-{}-{}-01", - readback["trace_id"].as_str().unwrap(), - readback["span_id"].as_str().unwrap() - )), - tracestate: Some("fixture=explicit".into()), - }; - let mut mismatch = json!({"input":[],"trace":carrier("12345678901234ab","01","fixture=other"),"tracing_context":token}); - let receipt = ops::prepare_submission_trace(&mut mismatch) - .unwrap() - .unwrap(); - assert_eq!(receipt["status"], "not_submitted"); - assert_eq!(receipt["reason"]["type"], "trace_context_conflict"); - assert!(codex_otel::sdk_trace::trace_span_context(token).is_ok()); - let mut matching = json!({"input":[],"trace":explicit,"tracing_context":token}); - assert!( - ops::prepare_submission_trace(&mut matching) + for flags in ["02", "03"] { + let explicit = W3cTraceContext { + traceparent: Some(format!( + "00-{}-{}-{flags}", + readback["trace_id"].as_str().unwrap(), + readback["span_id"].as_str().unwrap() + )), + tracestate: Some("fixture=explicit".into()), + }; + let mut mismatch = json!({"input":[],"trace":carrier("12345678901234ab","01","fixture=other"),"tracing_context":token}); + let receipt = ops::prepare_submission_trace(&mut mismatch) .unwrap() - .is_none() - ); - assert_eq!( - matching["trace"]["traceparent"], - explicit.traceparent.as_deref().unwrap() - ); - assert_eq!(matching["trace"]["tracestate"], "fixture=explicit"); - assert!(matching.get("tracing_context").is_none()); - let parsed = codex_otel::context_from_w3c_trace_context(&explicit).unwrap(); - assert!(codex_otel::sdk_trace::policy_for_context(&parsed).is_none()); - let _caller = codex_otel::sdk_trace::context(token).unwrap().attach(); - let restored = codex_otel::sdk_trace::current_scope_policy().unwrap(); - let original = codex_otel::sdk_trace::policy_for_context( - &codex_otel::sdk_trace::context(token).unwrap(), - ) - .unwrap(); - assert!(std::sync::Arc::ptr_eq(&restored, &original)); - assert!(restored.disabled); - assert!(!restored.include_sensitive_data); + .unwrap(); + assert_eq!(receipt["status"], "not_submitted"); + assert_eq!(receipt["reason"]["type"], "trace_context_conflict"); + assert!(codex_otel::sdk_trace::trace_span_context(token).is_ok()); + let mut matching = json!({"input":[],"trace":explicit,"tracing_context":token}); + assert!( + ops::prepare_submission_trace(&mut matching) + .unwrap() + .is_none() + ); + assert_eq!( + matching["trace"]["traceparent"], + explicit.traceparent.as_deref().unwrap() + ); + assert_eq!(matching["trace"]["tracestate"], "fixture=explicit"); + assert!(matching.get("tracing_context").is_none()); + let parsed = codex_otel::context_from_w3c_trace_context(&explicit).unwrap(); + assert!(codex_otel::sdk_trace::policy_for_context(&parsed).is_none()); + let _caller = codex_otel::sdk_trace::context(token).unwrap().attach(); + let restored = codex_otel::sdk_trace::current_scope_policy().unwrap(); + let original = codex_otel::sdk_trace::policy_for_context( + &codex_otel::sdk_trace::context(token).unwrap(), + ) + .unwrap(); + assert!(std::sync::Arc::ptr_eq(&restored, &original)); + assert!(restored.disabled); + assert!(!restored.include_sensitive_data); + } let mut scoped_only = json!({"input":[],"tracing_context":token}); assert!( ops::prepare_submission_trace(&mut scoped_only) diff --git a/packages/agents-codex/native/rust/native_interception.rs b/packages/agents-codex/native/rust/native_interception.rs index 846e5f84e..7a04d4e65 100644 --- a/packages/agents-codex/native/rust/native_interception.rs +++ b/packages/agents-codex/native/rust/native_interception.rs @@ -577,6 +577,7 @@ pub struct Control { native_ancestor: Option>, stopped: CancellationToken, creation_tracing: OnceLock>>, + creation_trace: OnceLock>, direct_turn: Mutex>>, code_mode_sessions: Mutex>>, pending_code_import: Mutex>>, @@ -624,6 +625,21 @@ impl Control { self.creation_tracing.get().cloned().flatten() } + pub(crate) fn creation_trace( + &self, + ) -> Result> { + match self.creation_tracing() { + Some(tracing) => codex_otel::sdk_trace::trace_carrier(tracing.token()) + .map(Some) + .map_err(anyhow::Error::msg), + None => Ok(self.raw_creation_trace()), + } + } + + pub(crate) fn raw_creation_trace(&self) -> Option { + self.creation_trace.get().cloned().flatten() + } + pub(crate) fn tracing_context(&self) -> codex_otel::sdk_trace::CapturedContext { if self.native_ancestor.is_some() { // A child created without tracing keeps that choice on later turns. @@ -725,6 +741,7 @@ impl Control { native_ancestor: None, stopped: CancellationToken::new(), creation_tracing: OnceLock::new(), + creation_trace: OnceLock::new(), direct_turn: Mutex::new(None), code_mode_sessions: Mutex::default(), pending_code_import: Mutex::default(), @@ -768,6 +785,7 @@ impl Control { native_ancestor: Some(Arc::downgrade(parent)), stopped: CancellationToken::new(), creation_tracing: OnceLock::new(), + creation_trace: OnceLock::new(), direct_turn: Mutex::new(None), code_mode_sessions: Mutex::default(), pending_code_import: Mutex::default(), @@ -2327,6 +2345,11 @@ impl ThreadLifecycleContributor for Boundary { ) -> ExtensionFuture<'a, ()> { Box::pin(async move { let creation_tracing = codex_otel::sdk_trace::CapturedContext::capture().context; + let creation_trace = if creation_tracing.is_none() { + codex_otel::current_span_w3c_trace_context() + } else { + None + }; let root = !matches!( input.session_source, SessionSource::Internal(_) | SessionSource::SubAgent(_) @@ -2359,6 +2382,7 @@ impl ThreadLifecycleContributor for Boundary { } }); let _ = control.creation_tracing.set(creation_tracing); + let _ = control.creation_trace.set(creation_trace); if strict_executor_writes { let _ = control.strict_executor_writes.set(()); } @@ -8221,6 +8245,43 @@ mod mcp_policy_tests { drop(output); drop(tx); } + #[test] + fn child_creation_trace_retains_raw_parent_without_sdk_policy() { + for flags in ["02", "03"] { + let original = codex_protocol::protocol::W3cTraceContext { + traceparent: Some(format!( + "00-11111111111111111111111111111111-2222222222222222-{flags}" + )), + tracestate: Some("tenant=original".into()), + }; + let attached = codex_otel::context_from_w3c_trace_context(&original) + .unwrap() + .attach(); + let control = Control::new("original-child".into(), false); + let _ = control + .creation_trace + .set(codex_otel::current_span_w3c_trace_context()); + assert!(control.creation_tracing().is_none()); + drop(attached); + let unrelated = codex_protocol::protocol::W3cTraceContext { + traceparent: Some("00-33333333333333333333333333333333-4444444444444444-01".into()), + tracestate: Some("tenant=unrelated".into()), + }; + let _attached = codex_otel::context_from_w3c_trace_context(&unrelated) + .unwrap() + .attach(); + let retained = control.creation_trace().unwrap().unwrap(); + assert_eq!(retained.traceparent, original.traceparent); + assert_eq!(retained.tracestate, original.tracestate); + assert!( + Control::new("another-child".into(), false) + .creation_trace() + .unwrap() + .is_none() + ); + } + } + #[tokio::test] async fn child_receipt_retains_original_tracing_after_origin_scope_ends() { const CHILD: &str = "AGENTS_CHILD_RECEIPT_TRACE_TEST"; @@ -8243,6 +8304,12 @@ mod mcp_policy_tests { let _ = control .creation_tracing .set(codex_otel::sdk_trace::CapturedContext::capture().context); + let _ = control + .creation_trace + .set(Some(codex_protocol::protocol::W3cTraceContext { + traceparent: Some("00-cccccccccccccccccccccccccccccccc-dddddddddddddddd-03".into()), + tracestate: None, + })); let mode = serde_json::from_value(serde_json::json!({"mode":"default", "settings":{"model":"test","reasoning_effort":null,"developer_instructions":null}})).unwrap(); crate::turn_settlement::Contributor .on_turn_start(TurnStartInput { @@ -8267,6 +8334,16 @@ mod mcp_policy_tests { let _unrelated = codex_otel::sdk_trace::context(unrelated_handle) .unwrap() .attach(); + assert!( + control + .creation_trace() + .unwrap() + .unwrap() + .traceparent + .as_deref() + .unwrap() + .starts_with("00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-") + ); for owner in [creation, tracing] { let snapshot: serde_json::Value = serde_json::from_str( &codex_otel::sdk_trace::trace_span_context(owner.token()).unwrap(), diff --git a/packages/agents-codex/native/rust/storage.rs b/packages/agents-codex/native/rust/storage.rs index 45a885679..3847299b4 100644 --- a/packages/agents-codex/native/rust/storage.rs +++ b/packages/agents-codex/native/rust/storage.rs @@ -74,6 +74,12 @@ pub struct MemoryThreadStore { pub(crate) completion_hold: std::sync::Mutex>>, #[cfg(test)] pub(crate) next_flush_hold: std::sync::Mutex>>, + #[cfg(test)] + pub(crate) completion_flush_hold: std::sync::Mutex>>, + #[cfg(test)] + pub(crate) flush_traces: std::sync::Mutex< + Option>>>, + >, inner: Arc, archived: Mutex>, // The native in-memory store applies this policy but does not expose it. @@ -90,6 +96,10 @@ impl Default for MemoryThreadStore { completion_hold: std::sync::Mutex::new(None), #[cfg(test)] next_flush_hold: std::sync::Mutex::new(None), + #[cfg(test)] + completion_flush_hold: std::sync::Mutex::new(None), + #[cfg(test)] + flush_traces: std::sync::Mutex::new(None), manual_writers: Mutex::default(), inner: Arc::new(InMemoryThreadStore::default()), archived: Mutex::new(HashMap::new()), @@ -548,14 +558,31 @@ impl ThreadStore for MemoryThreadStore { #[cfg(test)] { return Box::pin(async move { - let hold = { - let mut next = self.next_flush_hold.lock().unwrap(); + let completing = at_turn_completion_flush(); + if completing + && let Some(probe) = self.flush_traces.lock().unwrap().clone() + && let Some(trace) = codex_otel::current_span_w3c_trace_context() + { + probe.lock().unwrap().push((id, trace)); + } + let completion = if completing { + let mut next = self.completion_flush_hold.lock().unwrap(); if next.as_ref().is_some_and(|hold| hold.parent == id) { next.take() } else { None } + } else { + None }; + let hold = completion.or_else(|| { + let mut next = self.next_flush_hold.lock().unwrap(); + if next.as_ref().is_some_and(|hold| hold.parent == id) { + next.take() + } else { + None + } + }); if let Some(hold) = hold { hold.entered.add_permits(1); hold.release.acquire().await.unwrap().forget(); @@ -758,6 +785,32 @@ impl ThreadStore for MemoryThreadStore { } } +// Core's "turn" span exports as session_task.turn. Its completion flush runs +// after session_task.run returns and before publishing the terminal. +// Ignore startup and in-sample flushes in the after-last-sample test barrier. +#[cfg(test)] +fn at_turn_completion_flush() -> bool { + use tracing_subscriber::registry::LookupSpan; + tracing::Span::current() + .with_subscriber(|(id, dispatch)| { + let Some(registry) = dispatch.downcast_ref::() else { + return false; + }; + let Some(span) = registry.span(id) else { + return false; + }; + for span in span.scope() { + match span.metadata().name() { + "session_task.run" => return false, + "turn" => return true, + _ => {} + } + } + false + }) + .unwrap_or(false) +} + #[cfg(test)] pub(crate) struct CompletionHold { pub parent: ThreadId, diff --git a/packages/agents-codex/native/rust/subagents.rs b/packages/agents-codex/native/rust/subagents.rs index 4424b4e14..d0a91b86c 100644 --- a/packages/agents-codex/native/rust/subagents.rs +++ b/packages/agents-codex/native/rust/subagents.rs @@ -754,11 +754,7 @@ pub async fn submit( client_id: None, }); let creation_tracing = control.creation_tracing(); - if let Some(tracing) = &creation_tracing { - request.trace = Some( - codex_otel::sdk_trace::trace_carrier(tracing.token()).map_err(anyhow::Error::msg)?, - ); - } + request.trace = control.creation_trace()?; let settlement = thread .thread_extension_data() .get::() @@ -834,7 +830,8 @@ pub async fn submit( .as_ref() .map(|tracing| codex_otel::sdk_trace::trace_carrier(tracing.token())) .transpose() - .map_err(anyhow::Error::msg)?; + .map_err(anyhow::Error::msg)? + .or_else(|| control.raw_creation_trace()); if let TurnInput::UserInput { client_id, .. } = &mut request.input { *client_id = Some(steering_id.clone()); } @@ -945,6 +942,10 @@ pub async fn continue_retained( let Some(thread) = turn.thread.upgrade() else { return Ok(None); }; + let control = thread + .thread_extension_data() + .get::() + .context("Native child continuation lost its tracing owner")?; let settlement = thread .thread_extension_data() .get::() @@ -990,7 +991,8 @@ pub async fn continue_retained( .as_ref() .map(|tracing| codex_otel::sdk_trace::trace_carrier(tracing.token())) .transpose() - .map_err(anyhow::Error::msg)?; + .map_err(anyhow::Error::msg)? + .or_else(|| control.raw_creation_trace()); let seed = Arc::new(crate::turn_settlement::Seed::default()); init.insert(seed.clone()); diff --git a/packages/agents-codex/native/rust/turn_settlement_tests.rs b/packages/agents-codex/native/rust/turn_settlement_tests.rs index 2626bab49..8ea31dd57 100644 --- a/packages/agents-codex/native/rust/turn_settlement_tests.rs +++ b/packages/agents-codex/native/rust/turn_settlement_tests.rs @@ -1787,3 +1787,163 @@ async fn checkpoint_probe_cannot_hide_invalid_actual_instructions_with_a_prospec drop(_mutation);state.close().await.unwrap();engine.close().await.unwrap(); }).await.expect("invalid source instruction evidence escaped passive validation"); } + + +#[test] +fn raw_child_steering_and_retained_continuation_preserve_carrier() { + const CHILD: &str = "AGENTS_RAW_CHILD_CONTINUATION_TRACE_TEST"; + if std::env::var_os(CHILD).is_none() { + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "engine::settlement_tests::raw_child_steering_and_retained_continuation_preserve_carrier", "--nocapture"]) + .env(CHILD, "1").status().unwrap(); + assert!(status.success()); + return; + } + // This real Core fixture installs a process-global tracing subscriber and + // constructs large turn futures. Keep both isolated from other unit cases. + std::thread::Builder::new() + .name("raw-child-carrier".into()) + .stack_size(32 * 1024 * 1024) + .spawn(|| { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(raw_child_steering_and_retained_continuation_case()); + }) + .unwrap() + .join() + .unwrap(); +} + +async fn raw_child_steering_and_retained_continuation_case() { + use codex_otel::sdk_trace::FutureExt as _; + let telemetry_home = tempfile::tempdir().unwrap(); + crate::telemetry::initialize( + r#"{"otel.trace_processors":true,"analytics.enabled":false}"#, + &json!({"codex_home":telemetry_home.path(),"cwd":telemetry_home.path()}).to_string(), + ) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(30), async { + for flags in ["02", "03"] { + let provider = Provider::start().await; + let (_home, engine) = provider.engine().await; + let (mut startup, instructions) = engine.build_startup(serde_json::from_value(json!({"_managed_instructions":true})).unwrap()).await.unwrap(); + startup.thread_extension_init.insert(crate::native_agents::Policy { + roles: Vec::new(), max_depth: 1, inherited_instructions: None, + allowed: Default::default(), + }); + let parent = engine.start_prepared(startup, instructions, None).await.unwrap(); + let parent_state = parent.current().unwrap(); + let carrier = codex_protocol::protocol::W3cTraceContext { + traceparent: Some(format!("00-11111111111111111111111111111111-2222222222222222-{flags}")), + tracestate: Some("tenant=raw-child".into()), + }; + let source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: parent_state.id, depth: 1, + agent_path: Some(codex_protocol::AgentPath::root().join("raw_child").unwrap()), + agent_nickname: None, agent_role: None, + }); + let options = serde_json::from_value(json!({"session_source":source,"thread_source":ThreadSource::Subagent})).unwrap(); + let (startup, _) = engine.build_startup(options).await.unwrap(); + let child = engine.manager.start_thread(startup) + .with_context(codex_otel::context_from_w3c_trace_context(&carrier).unwrap()) + .await.unwrap(); + let child_id = child.thread_id; + { + let mut events = parent_state.events.lock().await; + parent_state.discover_children(&mut events).await.unwrap(); + } + let (instance, _, _) = crate::subagents::instance(&parent_state, child_id).await.unwrap().unwrap(); + let store = engine.store.as_any().downcast_ref::().unwrap(); + let traces = Arc::new(std::sync::Mutex::new(Vec::new())); + *store.flush_traces.lock().unwrap() = Some(traces.clone()); + let hold = Arc::new(CompletionHold { + parent: child_id, entered: Semaphore::new(0), release: Semaphore::new(0), + }); + *store.completion_flush_hold.lock().unwrap() = Some(hold.clone()); + let mut initial = request("initial child input"); + initial.trace = child.thread.thread_extension_data() + .get::().unwrap().creation_trace().unwrap(); + initial.start.final_output_json_schema = Some(json!({ + "type":"object", "properties":{"answer":{"type":"string"}}, + "required":["answer"], "additionalProperties":false, + })); + let seed = Arc::new(crate::turn_settlement::Seed::default()); + let mut init = child.thread.config_snapshot().await.turn_extension_init; + init.insert(seed.clone()); + let StartIfIdleSubmission::Started { turn_id } = child.thread.start_turn_if_idle( + codex_core_api::WithTurnExtensionData::new(initial, init) + ).await.unwrap() else { panic!("initial child was not admitted"); }; + seed.confirm(&turn_id); + let original = loop { + if let Some((_, _, Some(turn))) = crate::subagents::instance(&parent_state, child_id).await.unwrap() + && turn.turn_id == turn_id { + break turn; + } + parent_state.next_session_event().await.unwrap(); + }; + hold.entered.acquire().await.unwrap().forget(); + assert_eq!(provider.child_requests(child_id), 1); + let unrelated: Value = serde_json::from_str(&codex_otel::sdk_trace::start_trace_span( + r#"{"name":"unrelated","kind":"workflow","trace_id":"33333333333333333333333333333333","include_sensitive_data":true}"# + ).unwrap()).unwrap(); + let handle = unrelated["handle"].as_str().unwrap(); + let context = codex_otel::sdk_trace::context(handle).unwrap(); + let steered = crate::subagents::submit(&parent_state, child_id, instance, + json!({"input":[{"type":"text","text":"RAW_RETAINED_AFTER_LAST_SAMPLE"}]})) + .with_context(context.clone()).await.unwrap(); + assert!(Arc::ptr_eq(&original, &steered)); + assert_eq!(provider.child_requests(child_id), 1); + hold.release.add_permits(1); + loop { + let event = parent_state.next_session_event().await.unwrap(); + if event["subagent_id"] == child_id.to_string() + && matches!(event["msg"]["type"].as_str(), Some("task_complete" | "turn_complete")) { + break; + } + } + original.wait_with_usage().await.unwrap(); + assert_eq!(child.thread.thread_extension_data().get::().unwrap().observe().receipt.unwrap().unsampled_steering(), Some(true)); + let successor = crate::subagents::continue_retained(&parent_state, &original) + .with_context(context).await.unwrap().expect("real retained child successor"); + assert_ne!(successor.turn_id, original.turn_id); + loop { + let event = parent_state.next_session_event().await.unwrap(); + if event["subagent_id"] == child_id.to_string() + && event["msg"]["turn_id"] == successor.turn_id + && matches!(event["msg"]["type"].as_str(), Some("task_complete" | "turn_complete")) { + break; + } + } + successor.wait_with_usage().await.unwrap(); + assert_eq!(provider.child_requests(child_id), 2); + { + let requests = provider.requests.lock().unwrap(); + let requests: Vec<_> = requests.iter().filter(|request| { + let metadata: Value = serde_json::from_str(request["client_metadata"]["x-codex-turn-metadata"].as_str().unwrap()).unwrap(); + metadata["thread_id"] == child_id.to_string() + }).collect(); + assert!(!requests[0]["input"].to_string().contains("RAW_RETAINED_AFTER_LAST_SAMPLE")); + assert!(requests[1]["input"].to_string().contains("RAW_RETAINED_AFTER_LAST_SAMPLE")); + } + { + let observed = traces.lock().unwrap(); + let observed: Vec<_> = observed.iter().filter(|(id, _)| *id == child_id).collect(); + assert!(observed.len() >= 2); + let last = &observed.last().unwrap().1; + assert_eq!(&last.traceparent.as_deref().unwrap()[3..35], "11111111111111111111111111111111", "retained successor completion"); + for (_, trace) in observed { + assert_eq!(&trace.traceparent.as_deref().unwrap()[3..35], "11111111111111111111111111111111"); + assert!(trace.traceparent.as_deref().unwrap().ends_with(&format!("-{flags}"))); + assert_eq!(trace.tracestate, carrier.tracestate); + } + } + codex_otel::sdk_trace::end_trace_span(handle, r#"{"status":"ok"}"#).unwrap(); + codex_otel::sdk_trace::release_trace_span(handle); + parent_state.close().await.unwrap(); + engine.close().await.unwrap(); + } + }).await.expect("raw child retained continuation must settle"); +} diff --git a/packages/agents-codex/src/runtime/trace-context.ts b/packages/agents-codex/src/runtime/trace-context.ts index 8dad4eb1d..216e208d9 100644 --- a/packages/agents-codex/src/runtime/trace-context.ts +++ b/packages/agents-codex/src/runtime/trace-context.ts @@ -9,7 +9,7 @@ export interface TraceContextHeaders { /** * Immutable explicit native trace parent, independent of any tracing provider. * Traceparent accepts at most 512 ASCII characters, nonzero lowercase IDs, and - * version 00 flags 00 or 01. Future versions retain their opaque suffix. + * version 00 flags 00 through 03 (sampled and random trace ID). Future versions retain their opaque suffix. * Tracestate accepts 32 entries with keys and values up to 256 characters; * tenant system IDs accept at most 13 characters to match native OpenTelemetry. * This native profile does not accept every standards-valid W3C carrier. @@ -44,7 +44,8 @@ export class TraceContext { match[2] === '0'.repeat(32) || match[3] === '0'.repeat(16) || (match[1] === '00' && - (parent.length !== 55 || !['00', '01'].includes(match[4]!))) || + (parent.length !== 55 || + !['00', '01', '02', '03'].includes(match[4]!))) || (parent.length > 55 && parent[55] !== '-') || /[^\x21-\x7e]/.test(parent) ) diff --git a/packages/agents-codex/test/native/fixtures/trace-context-child.ts b/packages/agents-codex/test/native/fixtures/trace-context-child.ts index 2cb2cc24f..967cbe77a 100644 --- a/packages/agents-codex/test/native/fixtures/trace-context-child.ts +++ b/packages/agents-codex/test/native/fixtures/trace-context-child.ts @@ -4,6 +4,9 @@ import { z } from 'zod'; import { Agent, NativeProvider, + NativeAgents, + Environment, + fileSessionStore, TraceContext, telemetry, tool, @@ -11,6 +14,7 @@ import { type SessionState, } from '../../../src'; import { ModelServer, call, message } from './model-server'; +import { startPublicExecutor } from './public-executor'; const PARENT = '00-11111111111111111111111111111111-2222222222222222-01'; const OVERRIDE = '00-33333333333333333333333333333333-4444444444444444-01'; @@ -114,6 +118,286 @@ async function review() { } } +async function definedFlags(flags: string) { + let callback: ReturnType | undefined; + const server = new ModelServer([ + [call('observe', {}, { callId: 'flags-call' })], + [final('done')], + ]); + const agent = new Agent({ + instructions: 'Call observe.', + model: 'gpt-5', + provider: await providerFor(server), + tools: [ + tool({ + name: 'observe', + parameters: z.object({}), + execute: (_args, context) => { + assert(context.traceContext); + callback = context.traceContext.toHeaders(); + return 'accepted'; + }, + }), + ], + }); + const result = await agent.run('start', { + parentTrace: parent(PARENT.slice(0, -2) + flags), + }); + assert.equal(result.finalOutput, 'done'); + assert(callback); + server.assertExhausted(); + return { callback, turn: result.turn.turnId }; +} + +async function storeReopen(path: string) { + const callbacks: ReturnType[] = []; + const server = new ModelServer( + Array.from({ length: 3 }, (_, index) => [ + [call('observe', {}, { callId: `store-${index}` })], + [final('stored')], + ]).flat(), + ); + const agent = new Agent({ + instructions: 'Call observe.', + model: 'gpt-5', + provider: await providerFor(server), + tools: [ + tool({ + name: 'observe', + parameters: z.object({}), + execute: (_args, context) => { + assert(context.traceContext); + callbacks.push(context.traceContext.toHeaders()); + return 'accepted'; + }, + }), + ], + }); + const storage = fileSessionStore(path + '.store'); + let id: string | undefined; + const turns: string[] = []; + for (const carrier of [ + PARENT.slice(0, -2) + '03', + WORKER.slice(0, -2) + '03', + undefined, + ]) { + const session = await agent.session({ + storage, + resume: id, + parentTrace: carrier ? parent(carrier) : undefined, + }); + try { + id ??= session.id; + assert.equal(session.id, id); + const turn = await session.submit('observe after opening'); + assert.equal((await turn.wait()).finalOutput, 'stored'); + turns.push(turn.turnId!); + } finally { + await session.close(); + } + } + assert.equal(callbacks.length, 3); + server.assertExhausted(); + return { id, turns, callbacks }; +} + +async function ownedChildren(flags?: string, steering = false) { + const callbackTurns: string[] = []; + const callbacks: ReturnType[] = []; + let rootCallback: ReturnType | undefined; + const sampled = Promise.withResolvers(); + const releaseSample = Promise.withResolvers(); + const continued = Promise.withResolvers(); + const marker = 'RAW_CHILD_RETAINED_TRACE'; + const sampledMarker: boolean[] = []; + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + let workflow: Awaited> | undefined; + let traceId = flags ? PARENT.slice(3, 35) : undefined; + let policyChecks = 0; + const checkPolicy = () => { + if (flags) return; + const span = telemetry.getCurrentSpan(); + assert(span); + assert.equal(telemetry.getCurrentTrace(), workflow); + assert.equal(span.context.disabled, false); + assert.equal(span.context.includeSensitiveData, false); + policyChecks++; + }; + let rootId: string | undefined; + let roots = 0; + let children = 0; + const respond = async (request: Record) => { + const metadata = request.client_metadata as Record; + const id = String(JSON.parse(metadata['x-codex-turn-metadata']).thread_id); + rootId ??= id; + if (id === rootId) + return ++roots === 1 + ? [call('root_observe', {}, { callId: 'root-observe' })] + : roots === 2 + ? [ + call( + 'spawn_agent', + { task_name: 'trace_child', message: 'Call observe.' }, + { namespace: 'collaboration', callId: 'spawn-trace-child' }, + ), + ] + : [final('root done')]; + children++; + sampledMarker.push(JSON.stringify(request.input).includes(marker)); + if (steering && children === 2) { + sampled.resolve(); + await releaseSample.promise; + } + return children % 2 === 1 + ? [call('observe', {}, { callId: `child-observe-${children}` })] + : [final('child done')]; + }; + const server = new ModelServer(Array.from({ length: 7 }, () => respond)); + const agent = new Agent({ + instructions: 'Delegate.', + model: 'gpt-5.5', + provider: await providerFor(server), + tools: [ + new NativeAgents({ + beforeStep: () => { + checkPolicy(); + return undefined; + }, + }), + tool({ + name: 'root_observe', + parameters: z.object({}), + execute: (_args, context) => { + checkPolicy(); + assert(context.traceContext); + rootCallback = context.traceContext.toHeaders(); + return 'root observed'; + }, + }), + tool({ + name: 'observe', + parameters: z.object({}), + execute: async (_args, context) => { + checkPolicy(); + assert(context.traceContext); + callbacks.push(context.traceContext.toHeaders()); + callbackTurns.push(context.turnId); + if (callbacks.length === 2) continued.resolve(); + if (callbacks.length === 1) { + entered.resolve(); + await release.promise; + } + await Promise.resolve(); + checkPolicy(); + return 'accepted'; + }, + }), + ], + }); + const executor = await startPublicExecutor(); + try { + const open = async () => { + const session = await agent.session({ + ...(flags + ? { parentTrace: parent(PARENT.slice(0, -2) + flags) } + : { tracing: { includeSensitiveData: false } }), + environment: new Environment({ + url: executor.url, + cwd: executor.cwd, + headers: executor.headers, + tools: ['exec_command', 'write_stdin', 'apply_patch'], + approval: () => false, + }), + }); + try { + const root = await session.submit('delegate'); + assert.equal((await root.wait()).finalOutput, 'root done'); + return { session, root }; + } catch (error) { + release.resolve(); + await session.close(); + throw error; + } + }; + const { session, root } = flags + ? await open() + : await telemetry.withTrace( + { name: 'carrier.owner', includeSensitiveData: false }, + async (original) => { + workflow = original; + traceId = original.context.traceId; + return open(); + }, + ); + try { + const owned = await Array.fromAsync(session.subagents()); + assert.equal(owned.length, 1); + const child = owned[0]!; + assert(child.turn); + await Promise.race([ + entered.promise, + child.turn.wait().then(() => { + throw new Error('Child finished without entering observe'); + }), + ]); + release.resolve(); + let later; + if (steering) { + await sampled.promise; + const steered = await telemetry.withTrace( + { name: 'unrelated.steering', includeSensitiveData: true }, + () => child.submit(marker), + ); + assert.equal(steered, child.turn); + releaseSample.resolve(); + assert.equal((await child.turn.wait()).finalOutput, 'child done'); + await continued.promise; + for await (const event of session.events()) { + if ( + event.type === 'turn_finished' && + event.turnId === callbackTurns[1] + ) { + assert.equal(event.status, 'completed'); + break; + } + } + assert.deepEqual(sampledMarker.slice(0, 3), [false, false, true]); + } else { + assert.equal((await child.turn.wait()).finalOutput, 'child done'); + later = await telemetry.withTrace( + { name: 'unrelated.caller', includeSensitiveData: true }, + () => child.submit('call observe again'), + ); + } + if (later) assert.equal((await later.wait()).finalOutput, 'child done'); + if (steering) assert.deepEqual(sampledMarker, [false, false, true, true]); + assert.equal(callbacks.length, 2); + server.assertExhausted(); + return { + flags, + steering, + sampledMarker, + traceId, + rootCallback, + callbacks, + callbackTurns, + rootTurn: root.turnId, + childTurns: callbackTurns, + policyChecks, + }; + } finally { + release.resolve(); + releaseSample.resolve(); + await session.close(); + } + } finally { + release.resolve(); + releaseSample.resolve(); + await executor.close(); + } +} + async function graph() { const remoteServer = new ModelServer([[final('remote')]]); const remote = new Agent({ @@ -293,14 +577,25 @@ async function main() { traces: { kind: 'otlp-http', endpoint: collector!, protocol: 'json' }, }); try { - const result = - mode === 'graph' - ? await graph() - : mode === 'compact' - ? await compact() - : mode === 'review' - ? await review() - : await manual(mode!, path!); + const result = mode?.startsWith('flags-') + ? await definedFlags(mode.slice(6)) + : mode === 'store-reopen' + ? await storeReopen(path!) + : mode?.startsWith('owned-steering-') + ? await ownedChildren(mode.slice('owned-steering-'.length), true) + : mode?.startsWith('owned-children-') + ? await ownedChildren( + mode === 'owned-children-scoped' + ? undefined + : mode.slice('owned-children-'.length), + ) + : mode === 'graph' + ? await graph() + : mode === 'compact' + ? await compact() + : mode === 'review' + ? await review() + : await manual(mode!, path!); console.log(JSON.stringify(result)); } finally { await telemetry.shutdown(); diff --git a/packages/agents-codex/test/native/trace-context-native.test.ts b/packages/agents-codex/test/native/trace-context-native.test.ts index e4b8220c3..764a0f8e7 100644 --- a/packages/agents-codex/test/native/trace-context-native.test.ts +++ b/packages/agents-codex/test/native/trace-context-native.test.ts @@ -11,6 +11,8 @@ const execute = promisify(execFile); const PARENT = '00-11111111111111111111111111111111-2222222222222222-01'; const OVERRIDE = '00-33333333333333333333333333333333-4444444444444444-01'; const WORKER = '00-55555555555555555555555555555555-6666666666666666-01'; +const STATE = + 'tenant@abcdefghijklm=' + 'x'.repeat(256) + ',other=' + 'y'.repeat(256); type Span = { name: string; traceId: string; @@ -92,6 +94,67 @@ it('exports isolated root and compaction parents, queued snapshots, exact callba return false; }; try { + for (const flags of ['02', '03']) { + const receipt = await run('flags-' + flags); + expect(receipt.callback.traceparent.slice(3, 35)).toBe( + PARENT.slice(3, 35), + ); + expect(receipt.callback.traceparent.endsWith('-' + flags)).toBe(true); + if (flags === '03') + expectParent(dispatch(receipt.turn), PARENT.slice(0, -2) + flags); + } + for (const flags of ['02', '03']) { + for (const kind of ['owned-children-', 'owned-steering-']) { + const firstSpan = spans.length; + const owned = await run(kind + flags); + expect(owned.rootCallback.traceparent.slice(3, 35)).toBe( + PARENT.slice(3, 35), + ); + expect(owned.rootCallback.traceparent.endsWith('-' + flags)).toBe(true); + expect(owned.rootCallback.tracestate).toBe(STATE); + expect(owned.callbacks).toHaveLength(2); + for (const callback of owned.callbacks) { + expect(callback.traceparent.slice(3, 35)).toBe(PARENT.slice(3, 35)); + expect(callback.traceparent.endsWith('-' + flags)).toBe(true); + expect(callback.tracestate).toBe(STATE); + } + for (const span of spans + .slice(firstSpan) + .filter((span) => span.name.startsWith('op.dispatch.turn_input'))) + expect(span.traceId).toBe(PARENT.slice(3, 35)); + if (kind === 'owned-steering-') + expect(owned.sampledMarker).toEqual([false, false, true, true]); + if (flags === '03') { + const root = dispatch(owned.rootTurn); + expectParent(root, PARENT); + for (const turn of owned.childTurns) { + const child = dispatch(turn); + expect(child.traceId).toBe(root.traceId); + expect(descendsFrom(child, root)).toBe(true); + } + } + } + } + const scoped = await run('owned-children-scoped'); + expect(scoped.rootCallback.traceparent.slice(3, 35)).toBe(scoped.traceId); + expect(scoped.callbacks).toHaveLength(2); + expect(scoped.policyChecks).toBeGreaterThanOrEqual(4); + for (const callback of scoped.callbacks) + expect(callback.traceparent.slice(3, 35)).toBe(scoped.traceId); + const reopened = await run('store-reopen'); + expectParent(dispatch(reopened.turns[0]), PARENT); + expectParent(dispatch(reopened.turns[1]), WORKER); + for (const [index, expected] of [PARENT, WORKER].entries()) { + expect(reopened.callbacks[index].traceparent.slice(3, 35)).toBe( + expected.slice(3, 35), + ); + expect(reopened.callbacks[index].traceparent.endsWith('-03')).toBe(true); + } + const fresh = reopened.callbacks[2].traceparent; + expect([PARENT.slice(3, 35), WORKER.slice(3, 35)]).not.toContain( + fresh.slice(3, 35), + ); + expect(Number.parseInt(fresh.slice(-2), 16) & 2).toBe(0); const compact = await run('compact'); expectParent(dispatch(compact.first), OVERRIDE); expectParent(dispatch(compact.compact), PARENT); diff --git a/packages/agents-codex/test/trace-context.test.ts b/packages/agents-codex/test/trace-context.test.ts index 56375783f..259d002de 100644 --- a/packages/agents-codex/test/trace-context.test.ts +++ b/packages/agents-codex/test/trace-context.test.ts @@ -69,7 +69,17 @@ describe('explicit W3C trace carriers', () => { expect(new Headers(headers).get('tracestate')).toBeNull(); }); - it.each(['02', '09'])( + it.each(['00', '01', '02', '03'])( + 'round-trips defined trace flags %s', + (flags) => { + const value = parent.slice(0, -2) + flags; + expect(new TraceContext({ traceparent: value }).toHeaders()).toEqual({ + traceparent: value, + }); + }, + ); + + it.each(['04', '09'])( 'rejects trace flags %s outside the documented native profile', (flags) => { expect(