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
This commit is contained in:
Owen Lin
2026-07-25 16:06:54 +00:00
committed by copyberry
parent 3a08af44b2
commit af7f6f4d34
18 changed files with 183 additions and 8 deletions
@@ -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,
}));
@@ -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,
}))
}
@@ -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,
}),
);
@@ -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,
}))
}
+1
View File
@@ -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(
+22 -2
View File
@@ -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;
+41
View File
@@ -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;
@@ -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(
+17
View File
@@ -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<Instant>,
started_at_unix_secs: Option<i64>,
item_started_at_ms: HashMap<String, i64>,
first_token_at: Option<Instant>,
first_message_at: Option<Instant>,
}
@@ -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<i64> {
self.state.lock().await.item_started_at_ms.remove(item_id)
}
pub(crate) async fn complete_profile_and_duration_ms(
&self,
) -> (Option<i64>, Option<i64>, TurnProfile) {
+38
View File
@@ -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(
+30 -3
View File
@@ -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::<RolloutLine>)
.collect::<Result<Vec<_>, _>>()?
.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(())
}
+12
View File
@@ -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<i64>,
// 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::<ItemCompletedEvent>(value).unwrap();
assert_eq!(event.started_at_ms, None);
assert_eq!(event.completed_at_ms, 0);
}
@@ -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,
})),
];
+1
View File
@@ -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,
}));
+1
View File
@@ -1580,6 +1580,7 @@ mod tests {
client_id: None,
content: Vec::new(),
}),
started_at_ms: Some(0),
completed_at_ms: 1,
}));
store
@@ -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,
}))
}
@@ -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,
}))
}
@@ -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,
},
))])