diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index c79ef56be4..3a4a47ca99 100644 --- a/codex-rs/core/src/codex.rs +++ b/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, } -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 = None; + let mut inference_stream_recorded = false; let outcome: CodexResult = 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?; diff --git a/codex-rs/exec/src/event_processor_with_human_output.rs b/codex-rs/exec/src/event_processor_with_human_output.rs index b5c580ac47..e38fbeee2d 100644 --- a/codex-rs/exec/src/event_processor_with_human_output.rs +++ b/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, last_total_token_usage: Option, final_message: Option, + last_turn_started_at: Option, + last_turn_elapsed: Option, + last_turn_timing: Option, + turn_start_output_tokens: Option, + last_turn_output_tokens: Option, } 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 diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 70875377e9..0e4383fb28 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/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)] diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 66017a7e05..4656dce055 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/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) diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index 2c35cc0eef..d61469088c 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/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();