Refine turn timing with stream stats

codex-rs/core/src/codex.rs

@@ -203,6 +203,8 @@ use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
 use codex_protocol::config_types::WindowsSandboxLevel;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::DeveloperInstructions;
+use codex_protocol::models::ReasoningItemContent;
+use codex_protocol::models::ReasoningItemReasoningSummary;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::models::render_command_prefix_list;
@@ -469,6 +471,7 @@ pub(crate) struct TurnTimingState {
     tool_duration_nanos: AtomicU64,
     inference_calls: AtomicU64,
     response_wait_nanos: AtomicU64,
+    inference_stream_nanos: AtomicU64,
 }
 
 impl TurnTimingState {
@@ -484,6 +487,11 @@ impl TurnTimingState {
             .fetch_add(duration_nanos(duration), Ordering::Relaxed);
     }
 
+    pub(crate) fn record_inference_stream(&self, duration: std::time::Duration) {
+        self.inference_stream_nanos
+            .fetch_add(duration_nanos(duration), Ordering::Relaxed);
+    }
+
     pub(crate) fn snapshot(&self) -> TurnTimingStats {
         TurnTimingStats {
             tool_calls: self.tool_calls.load(Ordering::Relaxed),
@@ -494,6 +502,9 @@ impl TurnTimingState {
             response_wait_duration: std::time::Duration::from_nanos(
                 self.response_wait_nanos.load(Ordering::Relaxed),
             ),
+            inference_stream_duration: std::time::Duration::from_nanos(
+                self.inference_stream_nanos.load(Ordering::Relaxed),
+            ),
         }
     }
 }
@@ -550,6 +561,10 @@ impl TurnContext {
         self.timing.record_inference_wait(duration);
     }
 
+    pub(crate) fn record_inference_stream(&self, duration: std::time::Duration) {
+        self.timing.record_inference_stream(duration);
+    }
+
     pub(crate) fn timing_snapshot(&self) -> TurnTimingStats {
         self.timing.snapshot()
     }
@@ -3697,14 +3712,69 @@ struct SamplingRequestResult {
     last_agent_message: Option<String>,
 }
 
-fn response_wait_started(event: &ResponseEvent) -> bool {
-    !matches!(
-        event,
+fn response_stream_started(event: &ResponseEvent) -> bool {
+    match event {
         ResponseEvent::Created
-            | ResponseEvent::RateLimits(_)
-            | ResponseEvent::ModelsEtag(_)
-            | ResponseEvent::ServerReasoningIncluded(_)
-    )
+        | ResponseEvent::RateLimits(_)
+        | ResponseEvent::ModelsEtag(_)
+        | ResponseEvent::ServerReasoningIncluded(_)
+        | ResponseEvent::ReasoningSummaryPartAdded { .. }
+        | ResponseEvent::Completed { .. } => false,
+        ResponseEvent::OutputTextDelta(delta) => !delta.is_empty(),
+        ResponseEvent::ReasoningSummaryDelta { delta, .. } => !delta.is_empty(),
+        ResponseEvent::ReasoningContentDelta { delta, .. } => !delta.is_empty(),
+        ResponseEvent::OutputItemAdded(item) | ResponseEvent::OutputItemDone(item) => {
+            response_item_has_content(item)
+        }
+    }
+}
+
+fn response_item_has_content(item: &ResponseItem) -> bool {
+    match item {
+        ResponseItem::Message { content, .. } => content.iter().any(|entry| {
+            matches!(entry, ContentItem::OutputText { text } if !text.is_empty())
+        }),
+        ResponseItem::Reasoning {
+            summary,
+            content,
+            encrypted_content,
+            ..
+        } => {
+            let summary_has_text = summary.iter().any(|entry| {
+                matches!(entry, ReasoningItemReasoningSummary::SummaryText { text } if !text.is_empty())
+            });
+            let content_has_text = content.as_ref().is_some_and(|content| {
+                content.iter().any(|entry| {
+                    matches!(
+                        entry,
+                        ReasoningItemContent::ReasoningText { text }
+                            | ReasoningItemContent::Text { text }
+                            if !text.is_empty()
+                    )
+                })
+            });
+            let encrypted_has_text = encrypted_content
+                .as_ref()
+                .is_some_and(|text| !text.is_empty());
+            summary_has_text || content_has_text || encrypted_has_text
+        }
+        ResponseItem::LocalShellCall { .. } => true,
+        ResponseItem::FunctionCall { arguments, .. } => !arguments.is_empty(),
+        ResponseItem::FunctionCallOutput { output, .. } => {
+            !output.content.is_empty()
+                || output
+                    .content_items
+                    .as_ref()
+                    .is_some_and(|items| !items.is_empty())
+        }
+        ResponseItem::CustomToolCall { input, .. } => !input.is_empty(),
+        ResponseItem::CustomToolCallOutput { output, .. } => !output.is_empty(),
+        ResponseItem::WebSearchCall { action: Some(_), .. } => true,
+        ResponseItem::WebSearchCall { action: None, .. } => false,
+        ResponseItem::GhostSnapshot { .. } => true,
+        ResponseItem::Compaction { encrypted_content } => !encrypted_content.is_empty(),
+        ResponseItem::Other => true,
+    }
 }
 
 async fn drain_in_flight(
@@ -3809,6 +3879,8 @@ async fn try_run_sampling_request(
     let mut should_emit_turn_diff = false;
     let receiving_span = trace_span!("receiving_stream");
     let mut inference_wait_recorded = false;
+    let mut inference_stream_started_at: Option<std::time::Instant> = None;
+    let mut inference_stream_recorded = false;
     let outcome: CodexResult<SamplingRequestResult> = loop {
         let handle_responses = trace_span!(
             parent: &receiving_span,
@@ -3838,10 +3910,13 @@ async fn try_run_sampling_request(
             }
         };
 
-        if !inference_wait_recorded && response_wait_started(&event) {
+        if !inference_wait_recorded && response_stream_started(&event) {
             turn_context.record_inference_wait(response_wait_start.elapsed());
             sess.emit_turn_timing_update(&turn_context).await;
             inference_wait_recorded = true;
+            if inference_stream_started_at.is_none() {
+                inference_stream_started_at = Some(std::time::Instant::now());
+            }
         }
 
         sess.services
@@ -3901,6 +3976,13 @@ async fn try_run_sampling_request(
                 should_emit_turn_diff = true;
 
                 needs_follow_up |= sess.has_pending_input().await;
+                if !inference_stream_recorded {
+                    if let Some(start) = inference_stream_started_at {
+                        turn_context.record_inference_stream(start.elapsed());
+                        sess.emit_turn_timing_update(&turn_context).await;
+                        inference_stream_recorded = true;
+                    }
+                }
 
                 break Ok(SamplingRequestResult {
                     needs_follow_up,
@@ -3978,6 +4060,12 @@ async fn try_run_sampling_request(
         turn_context.record_inference_wait(response_wait_start.elapsed());
         sess.emit_turn_timing_update(&turn_context).await;
     }
+    if !inference_stream_recorded {
+        if let Some(start) = inference_stream_started_at {
+            turn_context.record_inference_stream(start.elapsed());
+            sess.emit_turn_timing_update(&turn_context).await;
+        }
+    }
 
     drain_in_flight(&mut in_flight, sess.clone(), turn_context.clone()).await?;
 

codex-rs/exec/src/event_processor_with_human_output.rs

@@ -30,6 +30,7 @@ use codex_core::protocol::StreamErrorEvent;
 use codex_core::protocol::TurnAbortReason;
 use codex_core::protocol::TurnCompleteEvent;
 use codex_core::protocol::TurnDiffEvent;
+use codex_core::protocol::TurnTimingStats;
 use codex_core::protocol::WarningEvent;
 use codex_core::protocol::WebSearchEndEvent;
 use codex_core::web_search::web_search_detail;
@@ -73,6 +74,11 @@ pub(crate) struct EventProcessorWithHumanOutput {
     last_message_path: Option<PathBuf>,
     last_total_token_usage: Option<codex_core::protocol::TokenUsageInfo>,
     final_message: Option<String>,
+    last_turn_started_at: Option<Instant>,
+    last_turn_elapsed: Option<std::time::Duration>,
+    last_turn_timing: Option<TurnTimingStats>,
+    turn_start_output_tokens: Option<i64>,
+    last_turn_output_tokens: Option<i64>,
 }
 
 impl EventProcessorWithHumanOutput {
@@ -99,6 +105,11 @@ impl EventProcessorWithHumanOutput {
                 last_message_path,
                 last_total_token_usage: None,
                 final_message: None,
+                last_turn_started_at: None,
+                last_turn_elapsed: None,
+                last_turn_timing: None,
+                turn_start_output_tokens: None,
+                last_turn_output_tokens: None,
             }
         } else {
             Self {
@@ -116,6 +127,11 @@ impl EventProcessorWithHumanOutput {
                 last_message_path,
                 last_total_token_usage: None,
                 final_message: None,
+                last_turn_started_at: None,
+                last_turn_elapsed: None,
+                last_turn_timing: None,
+                turn_start_output_tokens: None,
+                last_turn_output_tokens: None,
             }
         }
     }
@@ -244,7 +260,12 @@ impl EventProcessor for EventProcessorWithHumanOutput {
                 ts_msg!(self, "{}", message.style(self.dimmed));
             }
             EventMsg::TurnStarted(_) => {
-                // Ignore.
+                self.last_turn_started_at = Some(Instant::now());
+                self.turn_start_output_tokens = self
+                    .last_total_token_usage
+                    .as_ref()
+                    .map(|info| info.total_token_usage.output_tokens);
+                self.last_turn_output_tokens = None;
             }
             EventMsg::ElicitationRequest(ev) => {
                 ts_msg!(
@@ -260,7 +281,8 @@ impl EventProcessor for EventProcessorWithHumanOutput {
                 );
             }
             EventMsg::TurnComplete(TurnCompleteEvent {
-                last_agent_message, ..
+                last_agent_message,
+                timing,
             }) => {
                 let last_message = last_agent_message.as_deref();
                 if let Some(output_file) = self.last_message_path.as_deref() {
@@ -268,6 +290,18 @@ impl EventProcessor for EventProcessorWithHumanOutput {
                 }
 
                 self.final_message = last_agent_message;
+                self.last_turn_timing = timing;
+                self.last_turn_elapsed = self.last_turn_started_at.map(|start| start.elapsed());
+                self.last_turn_output_tokens = self
+                    .last_total_token_usage
+                    .as_ref()
+                    .map(|info| {
+                        let total = info.total_token_usage.output_tokens;
+                        match self.turn_start_output_tokens {
+                            Some(start) => total.saturating_sub(start),
+                            None => total,
+                        }
+                    });
 
                 return CodexStatus::InitiateShutdown;
             }
@@ -792,6 +826,58 @@ impl EventProcessor for EventProcessorWithHumanOutput {
             );
         }
 
+        if let Some(timing) = &self.last_turn_timing {
+            let tool_calls = timing.tool_calls;
+            let tool_label = if tool_calls == 1 { "call" } else { "calls" };
+            let tool_duration = format_duration(timing.local_tool_duration);
+
+            let inference_calls = timing.inference_calls;
+            let inference_label = if inference_calls == 1 {
+                "call"
+            } else {
+                "calls"
+            };
+            let response_wait = format_duration(timing.response_wait_duration);
+            let inference_stream = format_duration(timing.inference_stream_duration);
+
+            let mut parts = Vec::new();
+            if let Some(elapsed) = self.last_turn_elapsed {
+                parts.push(format!("Worked for {}", format_duration(elapsed)));
+            }
+            parts.push(format!(
+                "Local tools: {tool_calls} {tool_label}, {tool_duration}"
+            ));
+            parts.push(format!(
+                "Inference: {inference_calls} {inference_label}, {response_wait} wait, {inference_stream} stream"
+            ));
+            let summary = parts.join(" • ");
+            eprintln!("{summary}");
+        }
+
+        if let Some(output_tokens) = self.last_turn_output_tokens {
+            let output_tokens = output_tokens.max(0);
+            let mut detail = format!(
+                "Output tokens: {}",
+                format_with_separators(output_tokens)
+            );
+            if let Some(timing) = &self.last_turn_timing
+                && output_tokens > 0
+            {
+                let stream = timing.inference_stream_duration;
+                let per_token_ms = stream.as_secs_f64() * 1000.0 / output_tokens as f64;
+                let per_token = if per_token_ms >= 1000.0 {
+                    format!("{:.2}s/token", per_token_ms / 1000.0)
+                } else {
+                    format!("{:.2}ms/token", per_token_ms)
+                };
+                detail = format!(
+                    "{detail} • Stream: {} • {per_token}",
+                    format_duration(stream)
+                );
+            }
+            eprintln!("{detail}");
+        }
+
         // If the user has not piped the final message to a file, they will see
         // it twice: once written to stderr as part of the normal event
         // processing, and once here on stdout. We print the token summary above

codex-rs/protocol/src/protocol.rs

@@ -1109,6 +1109,9 @@ pub struct TurnTimingStats {
     pub local_tool_duration: Duration,
     #[ts(type = "string")]
     pub response_wait_duration: Duration,
+    #[serde(default)]
+    #[ts(type = "string")]
+    pub inference_stream_duration: Duration,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]

codex-rs/tui/src/chatwidget.rs

@@ -1783,6 +1783,9 @@ impl ChatWidget {
             response_wait_duration: current
                 .response_wait_duration
                 .saturating_sub(baseline.response_wait_duration),
+            inference_stream_duration: current
+                .inference_stream_duration
+                .saturating_sub(baseline.inference_stream_duration),
         };
         self.last_separator_timing = Some(current);
         Some(delta)

codex-rs/tui/src/history_cell.rs

@@ -1904,17 +1904,18 @@ impl HistoryCell for FinalMessageSeparator {
                 "calls"
             };
             let response_wait = format_duration(timing.response_wait_duration);
+            let inference_stream = format_duration(timing.inference_stream_duration);
 
             let summary = if let Some(elapsed_seconds) = self
                 .elapsed_seconds
                 .map(super::status_indicator_widget::fmt_elapsed_compact)
             {
                 format!(
-                    "─ Worked for {elapsed_seconds} • Local tools: {tool_calls} {tool_label}, {tool_duration} • Inference: {inference_calls} {inference_label}, {response_wait} wait ─"
+                    "─ Worked for {elapsed_seconds} • Local tools: {tool_calls} {tool_label}, {tool_duration} • Inference: {inference_calls} {inference_label}, {response_wait} wait, {inference_stream} stream ─"
                 )
             } else {
                 format!(
-                    "─ Local tools: {tool_calls} {tool_label}, {tool_duration} • Inference: {inference_calls} {inference_label}, {response_wait} wait ─"
+                    "─ Local tools: {tool_calls} {tool_label}, {tool_duration} • Inference: {inference_calls} {inference_label}, {response_wait} wait, {inference_stream} stream ─"
                 )
             };
             let summary_width = summary.width();