From af7f6f4d346a71e78869f4c0e507160f921fffed Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sat, 25 Jul 2026 15:59:00 +0000 Subject: [PATCH] Include item start times in completion events (#35363) ## What changed - Add an optional `started_at_ms` field to `ItemCompletedEvent`, retaining compatibility with older persisted events that lack it. - Track the first start timestamp for each in-flight item and attach it when emitting and persisting the completion event. If no start was recorded, use the completion timestamp as a fallback. - Emit a complete start/completion lifecycle for subagent activity items. ## Testing - Cover concurrent item timing, repeated starts, turn-boundary cleanup, the missing-start fallback, subagent activity, and persisted web-search events. GitOrigin-RevId: e7cec9c4f1ef6ba67f287e81fb4d7d856fcf87a7 --- .../src/protocol/thread_history.rs | 6 +++ .../thread_history_projection_tests.rs | 1 + .../app-server/src/bespoke_event_handling.rs | 3 ++ .../app-server/tests/suite/v2/thread_read.rs | 1 + codex-rs/core/src/agent/control_tests.rs | 1 + codex-rs/core/src/session/mod.rs | 24 ++++++++++- codex-rs/core/src/session/tests.rs | 41 +++++++++++++++++++ .../src/tools/handlers/multi_agents_v2.rs | 6 +-- codex-rs/core/src/turn_timing.rs | 17 ++++++++ codex-rs/core/src/turn_timing_tests.rs | 38 +++++++++++++++++ codex-rs/core/tests/suite/items.rs | 33 +++++++++++++-- codex-rs/protocol/src/protocol.rs | 12 ++++++ .../rollout/src/persistence_metrics_tests.rs | 3 ++ codex-rs/state/src/extract.rs | 1 + codex-rs/thread-store/src/local/mod.rs | 1 + .../src/local/model_context_tests.rs | 1 + .../thread_history_materialization_tests.rs | 1 + .../thread-store/src/thread_metadata_sync.rs | 1 + 18 files changed, 183 insertions(+), 8 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index a1c1cf66c9..90db564f82 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -1830,12 +1830,14 @@ mod tests { thread_id, turn_id: "turn-1".into(), item: entered, + started_at_ms: Some(0), completed_at_ms: 0, }), EventMsg::ItemCompleted(ItemCompletedEvent { thread_id, turn_id: "turn-1".into(), item: exited, + started_at_ms: Some(0), completed_at_ms: 0, }), EventMsg::TurnComplete(TurnCompleteEvent { @@ -2002,6 +2004,7 @@ mod tests { thread_id, turn_id: turn_id.to_string(), item: sleep_item, + started_at_ms: Some(0), completed_at_ms: 1_000, }), EventMsg::TurnComplete(TurnCompleteEvent { @@ -2056,6 +2059,7 @@ mod tests { saved_path: Some(saved_path.clone()), }, )), + started_at_ms: Some(0), completed_at_ms: 1_000, }), EventMsg::TurnComplete(TurnCompleteEvent { @@ -2123,6 +2127,7 @@ mod tests { thread_id, turn_id: turn_id.to_string(), item: command_item, + started_at_ms: Some(0), completed_at_ms: 1_000, }), EventMsg::ExecCommandEnd(ExecCommandEndEvent { @@ -4195,6 +4200,7 @@ mod tests { thread_id: ThreadId::new(), turn_id: "turn-a".into(), item: hook_prompt, + started_at_ms: Some(0), completed_at_ms: 0, })); diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs index 851b95ecbe..2eb7875deb 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs @@ -206,6 +206,7 @@ fn item_completed(thread_id: ThreadId, turn_id: &str, item: TurnItem) -> Rollout thread_id, turn_id: turn_id.to_string(), item, + started_at_ms: Some(0), completed_at_ms: 123, })) } diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index fac6bd5fd4..1f10da7ec4 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -3432,6 +3432,7 @@ mod tests { agent_path: AgentPath::try_from("/root/worker") .expect("agent path should parse"), }), + started_at_ms: Some(42), completed_at_ms: 42, }), }, @@ -3606,6 +3607,7 @@ mod tests { phase: None, memory_citation: None, }), + started_at_ms: Some(0), completed_at_ms: 0, }), ); @@ -3622,6 +3624,7 @@ mod tests { phase: None, memory_citation: None, }), + started_at_ms: Some(0), completed_at_ms: 0, }), ); diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 3c90f25683..a42a2a91c7 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -2170,6 +2170,7 @@ fn paginated_completed_item( thread_id, turn_id: turn_id.to_string(), item, + started_at_ms: Some(0), completed_at_ms: 1, })) } diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index ab14e8dbeb..7269561f88 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -944,6 +944,7 @@ async fn spawn_agent_fork_from_paginated_parent_uses_model_context_prefix() { client_id: None, content: Vec::new(), }), + started_at_ms: Some(0), completed_at_ms: 1, })), RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied( diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index d0a14d1223..e06dfb569f 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2092,13 +2092,17 @@ impl Session { } pub(crate) async fn emit_turn_item_started(&self, turn_context: &TurnContext, item: &TurnItem) { + let started_at_ms = turn_context + .turn_timing_state + .record_item_started(item.id(), now_unix_timestamp_ms()) + .await; self.send_event( turn_context, EventMsg::ItemStarted(ItemStartedEvent { thread_id: self.thread_id, turn_id: turn_context.sub_id.clone(), item: item.clone(), - started_at_ms: now_unix_timestamp_ms(), + started_at_ms, }), ) .await; @@ -2110,13 +2114,29 @@ impl Session { item: TurnItem, ) { record_turn_ttfm_metric(turn_context, &item).await; + let completed_at_ms = now_unix_timestamp_ms(); + let item_id = item.id(); + let started_at_ms = turn_context + .turn_timing_state + .take_item_started(&item_id) + .await + .unwrap_or_else(|| { + warn!( + thread_id = %self.thread_id, + turn_id = %turn_context.sub_id, + item_id = %item_id, + "item completed without a recorded start timestamp" + ); + completed_at_ms + }); self.send_event( turn_context, EventMsg::ItemCompleted(ItemCompletedEvent { thread_id: self.thread_id, turn_id: turn_context.sub_id.clone(), item, - completed_at_ms: now_unix_timestamp_ms(), + started_at_ms: Some(started_at_ms), + completed_at_ms, }), ) .await; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index e8bdeb72b9..87e9ec0600 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -2001,6 +2001,47 @@ async fn record_response_item_and_emit_turn_item_emits_hook_prompt_lifecycle() { assert!(rx.try_recv().is_err(), "no extra events expected"); } +#[tokio::test] +async fn item_completion_without_a_start_uses_completion_timestamp() { + let (session, turn_context, rx) = make_session_and_context_with_rx().await; + let item = TurnItem::UserMessage(UserMessageItem { + id: "missing-start".to_string(), + client_id: None, + content: Vec::new(), + }); + + session.emit_turn_item_completed(&turn_context, item).await; + + let completed = rx.recv().await.expect("completed item event"); + let EventMsg::ItemCompleted(event) = completed.msg else { + panic!("expected completed item event"); + }; + assert_eq!(event.started_at_ms, Some(event.completed_at_ms)); +} + +#[tokio::test] +async fn subagent_activity_emits_matching_start_and_completion() { + let (session, turn_context, rx) = make_session_and_context_with_rx().await; + let item = codex_protocol::items::SubAgentActivityItem { + id: "activity-1".to_string(), + kind: codex_protocol::protocol::SubAgentActivityKind::Started, + agent_thread_id: ThreadId::new(), + agent_path: AgentPath::root(), + }; + + crate::tools::handlers::multi_agents_v2::emit_sub_agent_activity(&session, &turn_context, item) + .await; + + let EventMsg::ItemStarted(started) = rx.recv().await.expect("started item event").msg else { + panic!("expected started item event"); + }; + let EventMsg::ItemCompleted(completed) = rx.recv().await.expect("completed item event").msg + else { + panic!("expected completed item event"); + }; + assert_eq!(completed.started_at_ms, Some(started.started_at_ms)); +} + #[tokio::test] async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() { let (mut session, turn_context) = make_session_and_context().await; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs index d1446ec621..7ee0205eca 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs @@ -46,9 +46,9 @@ pub(crate) async fn emit_sub_agent_activity( turn: &crate::session::turn_context::TurnContext, item: SubAgentActivityItem, ) { - session - .emit_turn_item_completed(turn, TurnItem::SubAgentActivity(item)) - .await; + let item = TurnItem::SubAgentActivity(item); + session.emit_turn_item_started(turn, &item).await; + session.emit_turn_item_completed(turn, item).await; } pub(super) fn communication_from_tool_message( diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index 0502ad6ee4..87f779867e 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::sync::Arc; use std::sync::Mutex as StdMutex; use std::time::Duration; @@ -49,6 +50,7 @@ pub(crate) struct TurnTimingState { struct TurnTimingStateInner { started_at: Option, started_at_unix_secs: Option, + item_started_at_ms: HashMap, first_token_at: Option, first_message_at: Option, } @@ -90,6 +92,7 @@ impl TurnTimingState { let mut state = self.state.lock().await; state.started_at = Some(started_at); state.started_at_unix_secs = Some(started_at_unix_ms / 1000); + state.item_started_at_ms.clear(); state.first_token_at = None; state.first_message_at = None; self.profile_state().start(started_at); @@ -100,6 +103,20 @@ impl TurnTimingState { self.state.lock().await.started_at_unix_secs } + pub(crate) async fn record_item_started(&self, item_id: String, started_at_ms: i64) -> i64 { + *self + .state + .lock() + .await + .item_started_at_ms + .entry(item_id) + .or_insert(started_at_ms) + } + + pub(crate) async fn take_item_started(&self, item_id: &str) -> Option { + self.state.lock().await.item_started_at_ms.remove(item_id) + } + pub(crate) async fn complete_profile_and_duration_ms( &self, ) -> (Option, Option, TurnProfile) { diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index e9748547fe..29eee14fc8 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -103,6 +103,44 @@ async fn turn_timing_state_records_turn_started_epoch_millis() { ); } +#[tokio::test] +async fn turn_timing_state_tracks_concurrent_items_and_preserves_first_start() { + let state = TurnTimingState::default(); + + state + .record_item_started("first".to_string(), /*started_at_ms*/ 100) + .await; + state + .record_item_started("second".to_string(), /*started_at_ms*/ 200) + .await; + assert_eq!( + state + .record_item_started("first".to_string(), /*started_at_ms*/ 300) + .await, + 100 + ); + + assert_eq!(state.take_item_started("second").await, Some(200)); + assert_eq!(state.take_item_started("first").await, Some(100)); +} + +#[tokio::test] +async fn turn_timing_state_preserves_in_flight_items_after_turn_completion() { + let state = TurnTimingState::default(); + state + .record_item_started("first".to_string(), /*started_at_ms*/ 100) + .await; + + state.mark_turn_started(Instant::now()).await; + assert_eq!(state.take_item_started("first").await, None); + + state + .record_item_started("second".to_string(), /*started_at_ms*/ 200) + .await; + let _ = state.complete_profile_and_duration_ms().await; + assert_eq!(state.take_item_started("second").await, Some(200)); +} + #[test] fn response_item_records_turn_ttft_for_first_output_signals() { assert!(response_item_records_turn_ttft( diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index c41c2bbf84..a0e9ca64fb 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -14,6 +14,9 @@ use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::ItemStartedEvent; use codex_protocol::protocol::Op; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::user_input::ByteRange; use codex_protocol::user_input::TextElement; use codex_protocol::user_input::UserInput; @@ -319,7 +322,10 @@ async fn web_search_item_is_emitted() -> anyhow::Result<()> { let server = start_mock_server().await; - let TestCodex { codex, .. } = test_codex().build(&server).await?; + let TestCodex { codex, .. } = test_codex() + .with_history_mode(ThreadHistoryMode::Paginated) + .build(&server) + .await?; let web_search_added = ev_web_search_call_added_partial("web-search-1", "in_progress"); let web_search_done = ev_web_search_call_done("web-search-1", "completed", "weather seattle"); @@ -362,9 +368,10 @@ async fn web_search_item_is_emitted() -> anyhow::Result<()> { let completed = wait_for_event_match(&codex, |ev| match ev { EventMsg::ItemCompleted(ItemCompletedEvent { item: TurnItem::WebSearch(item), + started_at_ms, completed_at_ms, .. - }) => Some((item.clone(), *completed_at_ms)), + }) => Some((item.clone(), *started_at_ms, *completed_at_ms)), _ => None, }) .await; @@ -373,7 +380,8 @@ async fn web_search_item_is_emitted() -> anyhow::Result<()> { assert_eq!(started.0.id, begin.call_id); assert!(started.1 > 0); assert_eq!(completed.0.id, begin.call_id); - assert!(completed.1 > 0); + assert_eq!(completed.1, Some(started.1)); + assert!(completed.2 > 0); assert_eq!( completed.0.action, WebSearchAction::Search { @@ -382,6 +390,25 @@ async fn web_search_item_is_emitted() -> anyhow::Result<()> { } ); + codex.flush_rollout().await?; + let rollout_path = codex.rollout_path().expect("paginated rollout path"); + let rollout = std::fs::read_to_string(rollout_path)?; + let persisted_completion = rollout + .lines() + .map(serde_json::from_str::) + .collect::, _>>()? + .into_iter() + .find_map(|line| match line.item { + RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) + if matches!(&event.item, TurnItem::WebSearch(item) if item.id == begin.call_id) => + { + Some(event) + } + _ => None, + }) + .expect("persisted web search completion"); + assert_eq!(persisted_completion.started_at_ms, Some(started.1)); + Ok(()) } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index ad5a646021..c692d040d2 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1839,6 +1839,9 @@ pub struct ItemCompletedEvent { pub thread_id: ThreadId, pub turn_id: String, pub item: TurnItem, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub started_at_ms: Option, // Old rollout files may contain ItemCompleted events for PlanItem without // this field. Default to 0 so those persisted rollouts still deserialize // after tightening the core event contract. @@ -5341,6 +5344,7 @@ mod tests { result: "Zm9v".into(), saved_path: Some(test_path_buf("/tmp/ig-1.png").abs()), }), + started_at_ms: Some(0), completed_at_ms: 0, }; @@ -5366,6 +5370,7 @@ mod tests { let event = ItemCompletedEvent { thread_id: ThreadId::new(), turn_id: "turn-1".into(), + started_at_ms: Some(0), completed_at_ms: 0, item: TurnItem::FileChange(FileChangeItem { id: "patch-1".into(), @@ -5404,6 +5409,7 @@ mod tests { let event = ItemCompletedEvent { thread_id: ThreadId::new(), turn_id: "turn-1".into(), + started_at_ms: Some(0), completed_at_ms: 0, item: TurnItem::McpToolCall(McpToolCallItem { id: "mcp-1".into(), @@ -5482,6 +5488,7 @@ mod tests { let completed = ItemCompletedEvent { thread_id: ThreadId::new(), turn_id: "turn-1".into(), + started_at_ms: Some(10), completed_at_ms: 20, item: TurnItem::CommandExecution(CommandExecutionItem { id: "exec-1".into(), @@ -5560,6 +5567,7 @@ mod tests { let completed = ItemCompletedEvent { thread_id: ThreadId::new(), turn_id: "turn-1".into(), + started_at_ms: Some(10), completed_at_ms: 20, item: TurnItem::DynamicToolCall(DynamicToolCallItem { id: "dynamic-1".into(), @@ -5604,6 +5612,7 @@ mod tests { let entered = ItemCompletedEvent { thread_id: ThreadId::new(), turn_id: "turn-1".into(), + started_at_ms: Some(0), completed_at_ms: 0, item: TurnItem::EnteredReviewMode(EnteredReviewModeItem { id: "entered-review".into(), @@ -5616,6 +5625,7 @@ mod tests { let exited = ItemCompletedEvent { thread_id: ThreadId::new(), turn_id: "turn-1".into(), + started_at_ms: Some(0), completed_at_ms: 0, item: TurnItem::ExitedReviewMode(ExitedReviewModeItem { id: "exited-review".into(), @@ -5676,12 +5686,14 @@ mod tests { thread_id: ThreadId::new(), turn_id: "turn-1".into(), item: TurnItem::UserMessage(UserMessageItem::new(&[])), + started_at_ms: None, completed_at_ms: 123, }) .unwrap(); value.as_object_mut().unwrap().remove("completed_at_ms"); let event = serde_json::from_value::(value).unwrap(); + assert_eq!(event.started_at_ms, None); assert_eq!(event.completed_at_ms, 0); } diff --git a/codex-rs/rollout/src/persistence_metrics_tests.rs b/codex-rs/rollout/src/persistence_metrics_tests.rs index 3a352263b7..46f70bdf84 100644 --- a/codex-rs/rollout/src/persistence_metrics_tests.rs +++ b/codex-rs/rollout/src/persistence_metrics_tests.rs @@ -258,6 +258,7 @@ fn item_completion_persistence_depends_on_history_mode() { client_id: None, content: Vec::new(), }), + started_at_ms: Some(0), completed_at_ms: 0, })); @@ -299,6 +300,7 @@ fn review_mode_persistence_depends_on_history_mode() { }, user_facing_hint: "Review requested.".to_string(), }), + started_at_ms: Some(0), completed_at_ms: 0, })), RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { @@ -308,6 +310,7 @@ fn review_mode_persistence_depends_on_history_mode() { id: "exited-review".to_string(), review_output: None, }), + started_at_ms: Some(0), completed_at_ms: 0, })), ]; diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 10ce38f6c3..7be5363760 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -255,6 +255,7 @@ mod tests { text: format!("{USER_MESSAGE_BEGIN} actual user request"), text_elements: Vec::new(), }])), + started_at_ms: Some(0), completed_at_ms: 0, })); diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 5d59666348..6f25492e9a 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -1580,6 +1580,7 @@ mod tests { client_id: None, content: Vec::new(), }), + started_at_ms: Some(0), completed_at_ms: 1, })); store diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index d85784c557..26dca54cb7 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -569,6 +569,7 @@ fn completed_user_message(turn_id: &str, message: &str) -> RolloutItem { text_elements: Vec::new(), }], }), + started_at_ms: Some(0), completed_at_ms: 0, })) } diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index 4d3da4d3b3..9ff102c3c3 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -1923,6 +1923,7 @@ fn completed_item(thread_id: ThreadId, turn_id: &str, item: TurnItem) -> Rollout thread_id, turn_id: turn_id.to_string(), item, + started_at_ms: Some(0), completed_at_ms: 1, })) } diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 44e502eacc..2cdd837580 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -519,6 +519,7 @@ mod tests { text: "first user text".to_string(), text_elements: Vec::new(), }])), + started_at_ms: Some(0), completed_at_ms: 0, }, ))])