Zheng-Lu · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026
@@ -84,6 +84,14 @@ jobs:
       - name: Test agentic CLI end to end
         run: python3 scripts/tests/agentic-cli-e2e-test.py
 
+      - name: Test the telemetry smoke checker
+        run: python3 -m unittest discover -s scripts/tests -p test_otel_fixtures.py
+
+      - name: Smoke-test telemetry through the local Collector
+        run: |
+          docker compose -f deploy/otel/docker-compose.yaml up -d
+          scripts/tests/otel-smoke.sh
+
   embedded-code-interpreter:
     name: Embedded code interpreter
     runs-on: ubuntu-latest

@@ -28,6 +28,10 @@
       "limit": 518,
       "reason": "Code-interpreter done-only ID backfill remains in the shared output-item slot validation path."
     },
+    "crates/agentic-server-core/src/executor/engine/multi_agent_tests.rs": {
+      "limit": 501,
+      "reason": "Declares the multi-agent execution-metrics test module (multi_agent_tests/metrics.rs). The module tree reached the 500-line limit with #432; the tests themselves live in the child file."
+    },
     "crates/agentic-server-core/src/tool/registry.rs": {
       "limit": 545,
       "reason": "The web_fetch executor binds through the existing registry validation and executor registration flow, like the code interpreter."

@@ -828,6 +828,18 @@ The boundary contract is one owner and one path per concern:
 - Delivery consumes translated frames and provides ordered, awaited client emission
   across inference rounds.
 
+Telemetry follows the same owners (`executor/telemetry/`; see
+[Observability](docs/deploying/observability.md)). The `ExecutionSpan` guard finalizes
+the `agentic.execute` span and the execution metrics together, exactly once, from `Drop`.
+Instruments come from `ExecutorMetrics` on `ExecutionContext`. The first-data timings
+share an `ExecutionClock`. The pipeline entry marks the first upstream line; multi-agent
+rounds run on their own pipelines, which share the execution's clock. The client-facing
+stream marks the first event and first output-text delta it yields to the transport, and
+its `InstrumentedStream` wrapper measures delivery wait. For retained (WebSocket)
+responses that stream is the event stream `ResponseEventSink::stream` returns. Stage timers wrap
+the code that owns each stage. A refactor that moves one of these boundaries must move its
+mark with it rather than add a second one.
+
 #### `accumulator/` — typed response assembly
 
 `ResponseAccumulator` owns validation, response lifecycle, typed output slots, delta

@@ -4,6 +4,8 @@ pub(crate) use estimate::{estimate_history_tokens, estimate_input_tokens};
 mod context;
 mod summary;
 
+use super::telemetry::FailureCategory;
+use super::telemetry::metrics::Stage;
 use super::telemetry::stages::CompactionTrigger;
 use context::item_has_meaningful_context;
 use summary::completed_summary_text;
@@ -122,6 +124,18 @@ async fn compact_items_with_trigger(
     exec_ctx: &ExecutionContext,
     auth: Option<&str>,
     trigger: CompactionTrigger,
+) -> ExecutorResult<(Vec<InputItem>, ResponseUsage, Option<String>)> {
+    let timer = exec_ctx.metrics.stage(Stage::Compaction);
+    let summarized = summarize_items(request, input, exec_ctx, auth).await;
+    timer.finish_result(&summarized);
+    summarized
+}
+
+async fn summarize_items(
+    request: &RequestPayload,
+    input: ResponsesInput,
+    exec_ctx: &ExecutionContext,
+    auth: Option<&str>,
 ) -> ExecutorResult<(Vec<InputItem>, ResponseUsage, Option<String>)> {
     let original_items = Vec::from(input);
     if !original_items.iter().any(item_has_meaningful_context) {
@@ -167,6 +181,7 @@ async fn compact_items_with_trigger(
     let mut agent = AgentPipeline::new(ctx, None, StreamRelay::detached());
     let response =
         fetch_blocking_payload(&mut agent, exec_ctx, auth, &crate::tool::ToolRegistry::default(), None).await?;
+    exec_ctx.metrics.record_response_usage(response.usage.as_ref());
     let summary = completed_summary_text(&response)?;
 
     Ok((
@@ -293,6 +308,7 @@ pub async fn compact_response(
 
     let response_id = ctx.response_id.clone();
     ctx.new_input_items.clone_from(&output);
+    let timer = exec_ctx.metrics.stage(Stage::Persist);
     match persist_prepared_turn(
         ctx,
         tool_search_metadata,
@@ -303,8 +319,13 @@ pub async fn compact_response(
     )
     .await
     {
-        Ok(()) | Err(ExecutorError::Storage(crate::StorageError::NotConfigured)) => {}
-        Err(error) => return Err(error),
+        Ok(()) => timer.finish(None),
+        // Without storage nothing was written, so there is no stage to time.
+        Err(ExecutorError::Storage(crate::StorageError::NotConfigured)) => timer.discard(),
+        Err(error) => {
+            timer.finish(Some(FailureCategory::from(&error)));
+            return Err(error);
+        }
     }
 
     Ok(CompactedResponse {

@@ -32,6 +32,8 @@ use crate::executor::request::{ExecutionContext, RequestContext};
 use crate::executor::response_budget::ExecutorResponseBudget;
 #[cfg(test)]
 use crate::executor::response_budget::MAX_EXECUTOR_RESPONSE_BYTES;
+use crate::executor::telemetry::Api;
+use crate::executor::telemetry::metrics::RoundCounter;
 use crate::tool::{ToolSearchMetadata, ToolSearchState};
 use crate::types::io::{InputItem, OutputItem, ResponseUsage, ResponsesInput};
 #[cfg(test)]
@@ -127,6 +129,7 @@ struct SingleAgentRun<'a> {
     response_budget: ExecutorResponseBudget,
     output: Vec<OutputItem>,
     usage: Option<ResponseUsage>,
+    rounds: RoundCounter,
 }
 
 impl<'a> SingleAgentRun<'a> {
@@ -139,6 +142,7 @@ impl<'a> SingleAgentRun<'a> {
             response_budget,
             output,
             usage: None,
+            rounds: exec_ctx.metrics.rounds(Api::Responses),
         })
     }
 
@@ -148,6 +152,7 @@ impl<'a> SingleAgentRun<'a> {
         stream_upstream: bool,
     ) -> ExecutorResult<(ResponsePayload, Option<ToolSearchMetadata>)> {
         loop {
+            self.rounds.begin_round();
             let RoundResult { mut payload, decision } = self
                 .root
                 .run_round(self.output.len(), auth, stream_upstream, &self.response_budget)
@@ -249,7 +254,15 @@ async fn run_blocking(
 
     let ch = exec_ctx.conv_handler.clone();
     let rh = exec_ctx.resp_handler.clone();
-    Box::pin(persist_if_needed(payload.clone(), ctx, tool_search_metadata, ch, rh)).await?;
+    Box::pin(persist_if_needed(
+        payload.clone(),
+        ctx,
+        tool_search_metadata,
+        ch,
+        rh,
+        &exec_ctx.metrics,
+    ))
+    .await?;
 
     Ok(payload)
 }

@@ -23,6 +23,7 @@ use crate::executor::rehydrate::prepare_reasoning_for_vllm;
 use crate::executor::relay::Release;
 use crate::executor::request::{ExecutionContext, RequestContext};
 use crate::executor::response_budget::ExecutorResponseBudget;
+use crate::executor::telemetry::metrics::Stage;
 use crate::executor::upstream::{fetch_blocking_payload, fetch_stream_payload};
 use crate::tool::{ToolRegistry, mcp, record_discovered_mcp_tools, responses_declarations};
 use crate::types::io::{InputItem, OutputItem, ResponsesInput, ToolChoice};
@@ -287,8 +288,9 @@ impl<'a> AgentTurn<'a> {
             compaction_usage.is_some(),
         )?;
         let round_span = crate::executor::telemetry::stages::inference_round(round);
-        let mut payload = if stream_upstream {
-            let payload = fetch_stream_payload(
+        let timer = self.exec_ctx.metrics.stage(Stage::Inference);
+        let fetched = if stream_upstream {
+            fetch_stream_payload(
                 self.pipeline,
                 self.exec_ctx,
                 auth,
@@ -297,11 +299,7 @@ impl<'a> AgentTurn<'a> {
                 response_budget,
             )
             .instrument(round_span)
-            .await?;
-            if round == 0 {
-                self.registry.clear_mcp_list_tool_items();
-            }
-            payload
+            .await
         } else {
             fetch_blocking_payload(
                 self.pipeline,
@@ -311,8 +309,16 @@ impl<'a> AgentTurn<'a> {
                 Some(response_budget),
             )
             .instrument(round_span)
-            .await?
+            .await
         };
+        timer.finish_result(&fetched);
+        let mut payload = fetched?;
+        if stream_upstream && round == 0 {
+            self.registry.clear_mcp_list_tool_items();
+        }
+        // Only what this round's upstream response reported; an absent
+        // `usage` records nothing.
+        self.exec_ctx.metrics.record_response_usage(payload.usage.as_ref());
         let mut round_usage = compaction_usage;
         accumulate_usage(&mut round_usage, payload.usage.take());
         payload.usage = round_usage;
@@ -404,14 +410,13 @@ impl<'a> AgentTurn<'a> {
         output_offset: usize,
         response_budget: &ExecutorResponseBudget,
     ) -> ExecutorResult<Vec<GatewayCallResult>> {
-        let policy = self.exec_ctx.gateway_scheduler_policy.clone();
         let (ctx, relay) = self.pipeline.parts_mut();
         if !relay.has_deferred() {
             return execute_and_emit_output_calls(
                 output_items,
                 &self.registry,
                 output_offset,
-                policy,
+                self.exec_ctx,
                 response_budget,
                 &mut self.tool_call_budget,
                 relay,
@@ -423,9 +428,10 @@ impl<'a> AgentTurn<'a> {
             output_items,
             &self.registry,
             output_offset,
-            policy,
+            self.exec_ctx.gateway_scheduler_policy.clone(),
             &mut self.tool_call_budget,
-        );
+        )
+        .with_metrics(&self.exec_ctx.metrics);
         let initial_event_run_len = scheduler.initial_event_run_len(output_items, &self.registry);
         emit_gateway_start_events(scheduler.event_plans().take(initial_event_run_len), relay).await?;
 

@@ -119,7 +119,7 @@ impl ExecuteRequest {
         let mut execution = self
             .execution
             .take()
-            .unwrap_or_else(|| ExecutionSpan::start(Api::Responses, Route::Executor, true));
+            .unwrap_or_else(|| ExecutionSpan::start(Api::Responses, Route::Executor, true, &self.exec_ctx.metrics));
         let limit = self.effective_max_stream_event_bytes();
         let prepared = if self.control.is_some() {
             Err(ExecutorError::InvalidRequest(
@@ -155,10 +155,14 @@ impl ExecuteRequest {
     /// # Errors
     /// Returns [`ExecutorError`] if rehydration or (non-streaming) LLM inference fails.
     pub async fn run(mut self) -> ExecutorResult<Either<ResponsePayload, BoxStream>> {
-        let execution = self
-            .execution
-            .take()
-            .unwrap_or_else(|| ExecutionSpan::start(Api::Responses, Route::Executor, self.payload.stream));
+        let execution = self.execution.take().unwrap_or_else(|| {
+            ExecutionSpan::start(
+                Api::Responses,
+                Route::Executor,
+                self.payload.stream,
+                &self.exec_ctx.metrics,
+            )
+        });
         let span = execution.span().clone();
         Box::pin(self.run_traced(execution)).instrument(span).await
     }

@@ -32,6 +32,10 @@ use crate::executor::{
     relay::{AgentFrame, AgentRoundId},
     request::ExecutionContext,
     response_budget::ExecutorResponseBudget,
+    telemetry::{
+        Api,
+        metrics::{ExecutionClock, RoundCounter},
+    },
 };
 use crate::tool::ToolSearchState;
 use crate::types::{
@@ -89,6 +93,10 @@ pub(super) struct MultiAgentRun {
     sealer: TranscriptSealer,
     payload: ResponsePayload,
     rounds: usize,
+    /// Every agent's upstream rounds, recorded once for the execution.
+    round_counter: RoundCounter,
+    /// The execution's timing state, shared with each round's pipeline.
+    clock: Option<ExecutionClock>,
     max_retained_bytes: usize,
     frame_sender: mpsc::Sender<AgentFrame>,
     frames: mpsc::Receiver<AgentFrame>,
@@ -173,6 +181,8 @@ impl MultiAgentRun {
             sealer: TranscriptSealer::new()?,
             payload,
             rounds: 0,
+            round_counter: exec.metrics.rounds(Api::Responses),
+            clock: pipeline.execution_clock().cloned(),
             max_retained_bytes: exec.responses_config.max_retained_bytes,
         })
     }

@@ -149,6 +149,9 @@ impl MultiAgentRun {
             StreamRelay::detached()
         };
         let mut agent = AgentPipeline::new(ctx, tool_search, relay);
+        if let Some(clock) = &self.clock {
+            agent.set_execution_clock(clock.clone());
+        }
         agent.set_agent_guidance(self.round_guidance(turn));
         let mut execution = context.execution.clone();
         let source = AgentRoundId {
@@ -181,6 +184,7 @@ impl MultiAgentRun {
             .set_phase(turn, AgentPhase::Inferring)
             .map_err(registry_error)?;
         self.rounds += 1;
+        self.round_counter.begin_round();
         Ok(())
     }
 

@@ -6,6 +6,8 @@ mod client_tools;
 mod explicit_compaction;
 #[path = "multi_agent_tests/live_control.rs"]
 mod live_control;
+#[path = "multi_agent_tests/metrics.rs"]
+mod metrics;
 #[path = "multi_agent_tests/root_completion.rs"]
 mod root_completion;
 #[path = "multi_agent_tests/service_tier.rs"]