From 277c597a4b50c3bbbbf9e7ce79004720b60a10d5 Mon Sep 17 00:00:00 2001 From: Josh Carver Date: Wed, 29 Jul 2026 12:49:34 -0700 Subject: [PATCH] feat(telemetry): name agent invocation spans --- crates/aether-core/src/core/agent_builder.rs | 2 +- crates/aether-core/src/core/agent_deps.rs | 6 ++-- crates/aether-core/src/events/observer.rs | 4 +-- .../aether-telemetry/src/genai_constants.rs | 1 + crates/aether-telemetry/src/otel_observer.rs | 13 ++++++-- .../aether-telemetry/src/telemetry_runtime.rs | 4 ++- .../aether-telemetry/tests/observer_tests.rs | 1 + .../tests/otlp_collector_tests.rs | 31 ++++++++++++++++--- 8 files changed, 50 insertions(+), 12 deletions(-) diff --git a/crates/aether-core/src/core/agent_builder.rs b/crates/aether-core/src/core/agent_builder.rs index 48ae47b9..095cca50 100644 --- a/crates/aether-core/src/core/agent_builder.rs +++ b/crates/aether-core/src/core/agent_builder.rs @@ -84,7 +84,7 @@ impl AgentBuilder { .context_window(spec.context_window) .model_settings(spec.model_settings.clone()); - if let Some(observer) = deps.observer() { + if let Some(observer) = deps.observer(&spec.name) { builder = builder.observer(observer); } diff --git a/crates/aether-core/src/core/agent_deps.rs b/crates/aether-core/src/core/agent_deps.rs index df92de3d..93438bc7 100644 --- a/crates/aether-core/src/core/agent_deps.rs +++ b/crates/aether-core/src/core/agent_deps.rs @@ -41,7 +41,9 @@ impl AgentDeps { } /// A fresh observer isolated to one agent, if a factory is configured. - pub fn observer(&self) -> Option> { - self.observer_factory.as_ref().map(|factory| factory.agent(self.parent_trace_context.as_ref())) + pub fn observer(&self, agent_name: &str) -> Option> { + self.observer_factory + .as_ref() + .map(|factory| factory.agent(Some(agent_name), self.parent_trace_context.as_ref())) } } diff --git a/crates/aether-core/src/events/observer.rs b/crates/aether-core/src/events/observer.rs index 2122a443..77ebcbe2 100644 --- a/crates/aether-core/src/events/observer.rs +++ b/crates/aether-core/src/events/observer.rs @@ -27,8 +27,8 @@ pub trait McpRequestInstrumentation: Send { /// Creates observers isolated to one agent or one inbound MCP request. pub trait ObserverFactory: Send + Sync { - /// A fresh observer for one agent, continuing `parent`'s trace when supplied. - fn agent(&self, parent: Option<&TraceContext>) -> Box; + /// A fresh observer for one agent, continuing the parent trace when supplied. + fn agent(&self, agent_name: Option<&str>, parent: Option<&TraceContext>) -> Box; /// Instrumentation for one inbound `tools/call` request. fn tool_call_request(&self, tool_name: &str, parent: Option<&TraceContext>) -> Box; diff --git a/crates/aether-telemetry/src/genai_constants.rs b/crates/aether-telemetry/src/genai_constants.rs index 827ece94..9564be74 100644 --- a/crates/aether-telemetry/src/genai_constants.rs +++ b/crates/aether-telemetry/src/genai_constants.rs @@ -8,6 +8,7 @@ pub use opentelemetry_semantic_conventions::SCHEMA_URL as GENAI_SEMCONV_SCHEMA_U use opentelemetry_semantic_conventions::{attribute, metric}; pub const ERROR_TYPE: &str = attribute::ERROR_TYPE; +pub const GEN_AI_AGENT_NAME: &str = attribute::GEN_AI_AGENT_NAME; pub const GEN_AI_INPUT_MESSAGES: &str = attribute::GEN_AI_INPUT_MESSAGES; pub const GEN_AI_OPERATION_NAME: &str = attribute::GEN_AI_OPERATION_NAME; pub const GEN_AI_OUTPUT_MESSAGES: &str = attribute::GEN_AI_OUTPUT_MESSAGES; diff --git a/crates/aether-telemetry/src/otel_observer.rs b/crates/aether-telemetry/src/otel_observer.rs index c77c5f05..3d442eb6 100644 --- a/crates/aether-telemetry/src/otel_observer.rs +++ b/crates/aether-telemetry/src/otel_observer.rs @@ -32,6 +32,7 @@ pub struct OtelInstrumentation { pub metrics: GenAiMetrics, pub capture_content: bool, pub root_parent: Option, + pub agent_name: Option, } impl OtelObserver { @@ -74,11 +75,19 @@ impl OtelObserver { let mut input = self.otel.capture().buffer(); input.set(&ContentBlock::join_text(content)); - let mut attributes = vec![KeyValue::new(semconv::GEN_AI_OPERATION_NAME, "invoke_agent")]; + let operation_name = "invoke_agent"; + let mut attributes = vec![KeyValue::new(semconv::GEN_AI_OPERATION_NAME, operation_name)]; + let span_name = match &self.otel.agent_name { + Some(agent_name) => { + attributes.push(KeyValue::new(semconv::GEN_AI_AGENT_NAME, agent_name.clone())); + format!("{operation_name} {agent_name}") + } + None => operation_name.to_string(), + }; if let Some(text) = input.get() { attributes.push(KeyValue::new(semconv::GEN_AI_INPUT_MESSAGES, input_messages_json(text))); } - let builder = SpanBuilder::from_name("invoke_agent").with_kind(SpanKind::Internal).with_attributes(attributes); + let builder = SpanBuilder::from_name(span_name).with_kind(SpanKind::Internal).with_attributes(attributes); let span_context = self.otel.start_span(builder, self.otel.root_parent.as_ref()); let span = SpanGuard::new(span_context, TURN_CANCEL_MESSAGE); self.turn = Some(TurnState::new(span, input, self.otel.capture())); diff --git a/crates/aether-telemetry/src/telemetry_runtime.rs b/crates/aether-telemetry/src/telemetry_runtime.rs index aa8e5a2d..68ae5d03 100644 --- a/crates/aether-telemetry/src/telemetry_runtime.rs +++ b/crates/aether-telemetry/src/telemetry_runtime.rs @@ -78,6 +78,7 @@ impl TelemetryRuntime { metrics: GenAiMetrics::new(&meter_provider.meter_with_scope(scope)), capture_content: config.capture_content, root_parent: trace_root.parent, + agent_name: None, }; Ok(Self { tracer_provider, meter_provider, instrumentation }) @@ -105,8 +106,9 @@ impl TelemetryRuntime { } impl ObserverFactory for OtelObserverFactory { - fn agent(&self, parent: Option<&TraceContext>) -> Box { + fn agent(&self, agent_name: Option<&str>, parent: Option<&TraceContext>) -> Box { let mut instrumentation = self.instrumentation.clone(); + instrumentation.agent_name = agent_name.map(str::to_string); if let Some(parent) = parent.and_then(extract_trace_context) { instrumentation.root_parent = Some(parent); } diff --git a/crates/aether-telemetry/tests/observer_tests.rs b/crates/aether-telemetry/tests/observer_tests.rs index 7f2ff281..1dcdb91f 100644 --- a/crates/aether-telemetry/tests/observer_tests.rs +++ b/crates/aether-telemetry/tests/observer_tests.rs @@ -507,6 +507,7 @@ impl OtelHarness { metrics, capture_content, root_parent: None, + agent_name: None, }); Self { observer: Some(observer), diff --git a/crates/aether-telemetry/tests/otlp_collector_tests.rs b/crates/aether-telemetry/tests/otlp_collector_tests.rs index 62efef61..9ba51843 100644 --- a/crates/aether-telemetry/tests/otlp_collector_tests.rs +++ b/crates/aether-telemetry/tests/otlp_collector_tests.rs @@ -113,6 +113,29 @@ async fn observer_factory_prefers_a_dynamic_parent_context() { assert_eq!(turn.parent_span_id, vec![0xfe, 0xdc, 0xba, 0x98, 0x76, 0x54, 0x32, 0x10]); } +#[tokio::test(flavor = "multi_thread")] +async fn observer_factory_names_agent_spans() { + let collector = FakeOtlpCollector::start().await; + let runtime = TelemetryRuntime::new(&TelemetryConfig { + headers: test_headers("named-agent"), + ..collector_config(&collector) + }) + .expect("runtime initializes against collector"); + { + let mut observer = runtime.observer_factory().agent(Some("PatchWaveFix"), None); + observer.on_event(&AgentEvent::Turn(TurnEvent::Started { content: vec![] })); + observer.on_event(&AgentEvent::Turn(TurnEvent::Ended { outcome: TurnOutcome::Completed })); + } + + runtime.shutdown().expect("runtime flushes traces"); + let exports = collector.exports(); + let spans = trace_spans(&exports); + let turn = spans.iter().find(|span| span.name == "invoke_agent PatchWaveFix").expect("named agent span exported"); + + assert_eq!(string_attribute(turn, "gen_ai.operation.name"), Some("invoke_agent")); + assert_eq!(string_attribute(turn, "gen_ai.agent.name"), Some("PatchWaveFix")); +} + #[tokio::test(flavor = "multi_thread")] async fn propagated_mcp_context_connects_parent_tool_server_and_child_agent_spans() { let collector = FakeOtlpCollector::start().await; @@ -123,7 +146,7 @@ async fn propagated_mcp_context_connects_parent_tool_server_and_child_agent_span .expect("runtime initializes against collector"); { let factory = runtime.observer_factory(); - let mut parent = factory.agent(None); + let mut parent = factory.agent(Some("PatchWaveFix"), None); parent.on_event(&AgentEvent::Turn(TurnEvent::Started { content: vec![] })); parent.on_event(&AgentEvent::Tool(ToolEvent::ExecutionStarted { tool_id: "call_1".to_string(), @@ -132,7 +155,7 @@ async fn propagated_mcp_context_connects_parent_tool_server_and_child_agent_span let outbound = parent.tool_trace_context("call_1").expect("outbound tool span provides propagation context"); let server = factory.tool_call_request("spawn_subagent", Some(&outbound)); - let mut child = factory.agent(server.trace_context().as_ref()); + let mut child = factory.agent(Some("Explore"), server.trace_context().as_ref()); child.on_event(&AgentEvent::Turn(TurnEvent::Started { content: vec![] })); child.on_event(&AgentEvent::Turn(TurnEvent::Ended { outcome: TurnOutcome::Completed })); server.finish(None); @@ -149,7 +172,7 @@ async fn propagated_mcp_context_connects_parent_tool_server_and_child_agent_span let server = spans.iter().find(|span| span.name == "tools/call spawn_subagent").expect("MCP server span exported"); let child = spans .iter() - .find(|span| span.name == "invoke_agent" && span.parent_span_id == server.span_id) + .find(|span| span.name == "invoke_agent Explore" && span.parent_span_id == server.span_id) .expect("child agent span is parented by the MCP server span"); assert_eq!(server.trace_id, tool.trace_id); @@ -298,7 +321,7 @@ fn assert_propagated_hierarchy(spans: &[&Span], expected_trace_id: &[u8], expect /// Feeds one complete turn through a fresh observer, whose spans the runtime /// exports once the observer is dropped. fn observe_a_turn(runtime: &TelemetryRuntime, parent: Option<&TraceContext>) { - let mut observer = runtime.observer_factory().agent(parent); + let mut observer = runtime.observer_factory().agent(None, parent); for event in events() { observer.on_event(&event); }