diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 414537133b..53d7404eae 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs b/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs index 7bcc341c02..0a25728b81 100644 --- a/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs @@ -109,6 +109,7 @@ fn guardian_stdin_reviews_preserve_parent_command_history() { let start_turn = |turn_id: &str| { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, 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 9485e9e8f5..75dacd5dc8 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -104,6 +104,7 @@ pub struct ThreadHistoryItemChange { #[derive(Debug, Clone, PartialEq)] pub struct ThreadHistoryTurnChange { pub turn_id: String, + pub root_turn_id: Option, pub status: TurnStatus, pub error: Option, pub started_at: Option, @@ -131,6 +132,7 @@ impl ThreadHistoryTurnChange { fn from_pending_turn(turn: &PendingTurn) -> Self { Self { turn_id: turn.id.clone(), + root_turn_id: turn.root_turn_id.clone(), status: turn.status.clone(), error: turn.error.clone(), started_at: turn.started_at, @@ -1242,11 +1244,12 @@ impl ThreadHistoryBuilder { fn handle_turn_started(&mut self, payload: &TurnStartedEvent) { self.finish_current_turn(); - let turn = self + let mut turn = self .new_turn(Some(payload.turn_id.clone())) .with_status(TurnStatus::InProgress) .with_started_at(payload.started_at) .opened_explicitly(); + turn.root_turn_id = payload.root_turn_id.clone(); self.record_changed_pending_turn(&turn); self.current_turn = Some(turn); } @@ -1362,6 +1365,7 @@ impl ThreadHistoryBuilder { }); PendingTurn { id, + root_turn_id: None, items: Vec::new(), item_index: TurnItemIndex::default(), error: None, @@ -1593,6 +1597,7 @@ impl TurnItemIndex { struct PendingTurn { id: String, + root_turn_id: Option, items: Vec, item_index: TurnItemIndex, error: Option, @@ -2033,6 +2038,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2098,6 +2104,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2166,6 +2173,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2259,6 +2267,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2376,6 +2385,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2476,6 +2486,7 @@ mod tests { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-image".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2869,6 +2880,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2936,6 +2948,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3071,6 +3084,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3150,6 +3164,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3234,6 +3249,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3330,6 +3346,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3424,6 +3441,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3495,6 +3513,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3519,6 +3538,7 @@ mod tests { }), EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-b".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3601,6 +3621,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3625,6 +3646,7 @@ mod tests { }), EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-b".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3701,6 +3723,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3769,6 +3792,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3837,6 +3861,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3861,6 +3886,7 @@ mod tests { }), EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-b".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3917,6 +3943,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, @@ -3932,6 +3959,7 @@ mod tests { }), EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-b".into(), + root_turn_id: None, trace_id: None, started_at: Some(30), model_context_window: None, @@ -4019,6 +4047,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4043,6 +4072,7 @@ mod tests { }), EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-b".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4089,6 +4119,7 @@ mod tests { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-compact".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4361,6 +4392,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4423,6 +4455,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4482,6 +4515,7 @@ mod tests { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, @@ -4552,6 +4586,7 @@ mod tests { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4612,6 +4647,7 @@ mod tests { let mut builder = ThreadHistoryBuilder::new(); builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4640,6 +4676,7 @@ mod tests { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4687,6 +4724,7 @@ mod tests { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4752,6 +4790,7 @@ mod tests { }], changed_turns: vec![ThreadHistoryTurnChange { turn_id: "rollout-0".into(), + root_turn_id: None, status: TurnStatus::Completed, error: None, started_at: None, @@ -4846,6 +4885,7 @@ mod tests { let start_changes = builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg( EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: Some("root-turn".into()), trace_id: None, started_at: Some(10), model_context_window: None, @@ -4858,6 +4898,7 @@ mod tests { changed_items: Vec::new(), changed_turns: vec![ThreadHistoryTurnChange { turn_id: "turn-a".into(), + root_turn_id: Some("root-turn".into()), status: TurnStatus::InProgress, error: None, started_at: Some(10), @@ -4896,6 +4937,7 @@ mod tests { changed_items: Vec::new(), changed_turns: vec![ThreadHistoryTurnChange { turn_id: "turn-a".into(), + root_turn_id: Some("root-turn".into()), status: TurnStatus::Completed, error: None, started_at: Some(10), @@ -4943,6 +4985,7 @@ mod tests { }], changed_turns: vec![ThreadHistoryTurnChange { turn_id: "rollout-0".into(), + root_turn_id: None, status: TurnStatus::Completed, error: None, started_at: None, @@ -4960,6 +5003,7 @@ mod tests { let changes = builder.handle_rollout_items_with_changes(&[ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: Some("root-turn".into()), trace_id: None, started_at: Some(10), model_context_window: None, @@ -4982,6 +5026,7 @@ mod tests { changed_items: Vec::new(), changed_turns: vec![ThreadHistoryTurnChange { turn_id: "turn-a".into(), + root_turn_id: Some("root-turn".into()), status: TurnStatus::Completed, error: None, started_at: Some(10), @@ -4999,6 +5044,7 @@ mod tests { let changes = builder.handle_rollout_items_with_changes(&[ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-a".into(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs index f5a1e238f5..a0b2a14ac7 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs @@ -23,6 +23,7 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet { RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => ThreadHistoryChangeSet { changed_turns: vec![ThreadHistoryTurnChange { turn_id: event.turn_id.clone(), + root_turn_id: event.root_turn_id.clone(), status: TurnStatus::InProgress, error: None, started_at: event.started_at, @@ -34,6 +35,7 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet { RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => ThreadHistoryChangeSet { changed_turns: vec![ThreadHistoryTurnChange { turn_id: event.turn_id.clone(), + root_turn_id: None, status: if event.error.is_some() { TurnStatus::Failed } else { @@ -58,6 +60,7 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet { ThreadHistoryChangeSet { changed_turns: vec![ThreadHistoryTurnChange { turn_id: turn_id.clone(), + root_turn_id: None, status: TurnStatus::Interrupted, error: None, started_at: event.started_at, 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 f4abca35cf..4210011287 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 @@ -27,6 +27,7 @@ fn projects_turn_lifecycle_without_prior_builder_state() { let started = project(RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: Some("root-turn".into()), trace_id: None, started_at: Some(10), model_context_window: None, @@ -49,11 +50,16 @@ fn projects_turn_lifecycle_without_prior_builder_state() { assert_eq!(started.changed_turns[0].turn_id, "turn-1"); assert_eq!(started.changed_turns[0].status, TurnStatus::InProgress); assert_eq!(started.changed_turns[0].started_at, Some(10)); + assert_eq!( + started.changed_turns[0].root_turn_id.as_deref(), + Some("root-turn") + ); assert_eq!( completed, ThreadHistoryChangeSet { changed_turns: vec![ThreadHistoryTurnChange { turn_id: "turn-1".to_string(), + root_turn_id: None, status: TurnStatus::Completed, error: None, started_at: Some(10), @@ -90,6 +96,7 @@ fn projects_failed_turn_completion_as_snapshot() { ThreadHistoryChangeSet { changed_turns: vec![ThreadHistoryTurnChange { turn_id: "turn-1".to_string(), + root_turn_id: None, status: TurnStatus::Failed, error: Some(TurnError { misalignment: None, @@ -240,6 +247,7 @@ fn projects_identified_turn_aborts() { ThreadHistoryChangeSet { changed_turns: vec![ThreadHistoryTurnChange { turn_id: "turn-1".to_string(), + root_turn_id: None, status: TurnStatus::Interrupted, error: None, started_at: Some(10), diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index be146d82c7..b5171d2320 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -3281,6 +3281,7 @@ mod tests { "turn-1", &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: Some(42), model_context_window: None, @@ -3316,6 +3317,7 @@ mod tests { id: "turn-1".to_string(), msg: EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: Some(42), model_context_window: None, @@ -3592,6 +3594,7 @@ mod tests { &event_turn_id, &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: event_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: Some(42), model_context_window: None, diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 875306fdb4..818ed4ba95 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -1194,6 +1194,7 @@ mod thread_processor_behavior_tests { "turn-1", &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/app-server/src/turn_cost_worker_tests.rs b/codex-rs/app-server/src/turn_cost_worker_tests.rs index c1b62699f7..efd940492c 100644 --- a/codex-rs/app-server/src/turn_cost_worker_tests.rs +++ b/codex-rs/app-server/src/turn_cost_worker_tests.rs @@ -88,6 +88,7 @@ async fn handle_observes_only_matching_model_provider() { id: "turn-1".to_string(), msg: EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index d19eedfdcc..0253646ae5 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -1102,6 +1102,7 @@ async fn thread_fork_can_cut_before_unfinished_stored_turn() -> Result<()> { &source_path, &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: unfinished_turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -1376,6 +1377,7 @@ async fn thread_fork_creates_reference_backed_paginated_thread() -> Result<()> { for item in [ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, @@ -1705,6 +1707,7 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted( source_path.as_path(), &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "active-turn".to_string(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, 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 48633ce600..85c7a1e86a 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -2222,6 +2222,7 @@ async fn read_items_page( fn paginated_turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index e977da7003..69da60d821 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -3866,6 +3866,7 @@ async fn cold_paginated_resume_restores_usage_without_loading_turns() -> Result< &path, &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: canonical_turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -3977,6 +3978,7 @@ async fn cold_paginated_resume_omits_usage_when_its_turn_is_ambiguous() -> Resul &path, &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: interrupted_turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4119,6 +4121,7 @@ async fn thread_resume_token_usage_replay_ignores_stale_interrupted_tail_turn() "type": "event_msg", "payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent { turn_id: stale_turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4207,6 +4210,7 @@ async fn thread_resume_token_usage_replay_can_belong_to_interrupted_turn() -> Re "type": "event_msg", "payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent { turn_id: interrupted_turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -4520,6 +4524,7 @@ async fn thread_resume_and_read_interrupt_incomplete_rollout_turn_when_thread_is "type": "event_msg", "payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs index 6cc196667b..43aea1dd62 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs @@ -97,6 +97,7 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() -> }), RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index d2a2010b5f..f8cb0e5457 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -476,6 +476,7 @@ async fn get_status_returns_not_found_without_manager() { async fn on_event_updates_status_from_task_started() { let status = agent_status_from_event(&EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 55cc1718ba..28bd8bf052 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -47,7 +47,6 @@ use codex_protocol::models::InternalChatMessageMetadataPassthrough; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::WarningEvent; use codex_protocol::user_input::UserInput; use codex_rollout_trace::InferenceTraceContext; @@ -149,14 +148,7 @@ pub(crate) async fn run_compact_task( turn_context: Arc, input: Vec, ) -> CodexResult<()> { - let start_event = EventMsg::TurnStarted(TurnStartedEvent { - turn_id: turn_context.sub_id.clone(), - trace_id: turn_context.trace_id.clone(), - started_at: turn_context.turn_timing_state.started_at_unix_secs().await, - model_context_window: turn_context.model_context_window(), - collaboration_mode_kind: turn_context.mode(), - }); - sess.send_event(&turn_context, start_event).await; + sess.emit_turn_started(&turn_context).await; run_compact_task_inner( sess.clone(), turn_context, diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 376ad06c99..7a5d3ad416 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -49,7 +49,6 @@ use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TruncationPolicy; -use codex_protocol::protocol::TurnStartedEvent; use codex_rollout_trace::CompactionCheckpointTracePayload; use codex_rollout_trace::InferenceTraceContext; use codex_utils_output_truncation::approx_token_count; @@ -111,14 +110,7 @@ pub(crate) async fn run_remote_compact_task( let step_context = sess .capture_step_context(Arc::clone(&turn_context), &CancellationToken::new()) .await?; - let start_event = EventMsg::TurnStarted(TurnStartedEvent { - turn_id: turn_context.sub_id.clone(), - trace_id: turn_context.trace_id.clone(), - started_at: turn_context.turn_timing_state.started_at_unix_secs().await, - model_context_window: turn_context.model_context_window(), - collaboration_mode_kind: turn_context.mode(), - }); - sess.send_event(&turn_context, start_event).await; + sess.emit_turn_started(&turn_context).await; let compaction_metadata = CompactionTurnMetadata::new( CompactionTrigger::Manual, diff --git a/codex-rs/core/src/compact_token_budget.rs b/codex-rs/core/src/compact_token_budget.rs index 60d25c53d8..3e6ce1beaa 100644 --- a/codex-rs/core/src/compact_token_budget.rs +++ b/codex-rs/core/src/compact_token_budget.rs @@ -14,8 +14,6 @@ use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; use codex_protocol::items::ContextCompactionItem; use codex_protocol::items::TurnItem; -use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::TurnStartedEvent; use tokio_util::sync::CancellationToken; /// Runs token-budget manual compaction as a normal compaction lifecycle. @@ -27,14 +25,7 @@ pub(crate) async fn run_manual_compact_task( sess: Arc, turn_context: Arc, ) -> CodexResult<()> { - let start_event = EventMsg::TurnStarted(TurnStartedEvent { - turn_id: turn_context.sub_id.clone(), - trace_id: turn_context.trace_id.clone(), - started_at: turn_context.turn_timing_state.started_at_unix_secs().await, - model_context_window: turn_context.model_context_window(), - collaboration_mode_kind: turn_context.mode(), - }); - sess.send_event(&turn_context, start_event).await; + sess.emit_turn_started(&turn_context).await; // Manual compaction runs outside run_turn, so it captures its own current step. let step_context = sess diff --git a/codex-rs/core/src/realtime_history_tests.rs b/codex-rs/core/src/realtime_history_tests.rs index 48db0239b3..55e0019b21 100644 --- a/codex-rs/core/src/realtime_history_tests.rs +++ b/codex-rs/core/src/realtime_history_tests.rs @@ -29,6 +29,7 @@ fn started_state() -> RealtimeHistoryState { state.observe(&EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -128,6 +129,7 @@ fn interrupted_turn_is_not_associated_with_a_new_voice_session( state.observe(&EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 98bfa792b2..cb51fdfb19 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -144,6 +144,7 @@ use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::TurnContextNetworkItem; use codex_protocol::protocol::TurnEnvironmentSelection; +use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::W3cTraceContext; use codex_protocol::protocol::WorldStateItem; use codex_protocol::request_permissions::PermissionGrantScope; @@ -2184,6 +2185,24 @@ impl Session { .get_or_init(GuardianReviewSessionManager::default) } + pub(crate) async fn emit_turn_started(&self, turn_context: &TurnContext) { + let event = TurnStartedEvent { + turn_id: turn_context.sub_id.clone(), + root_turn_id: Some( + turn_context + .turn_metadata_state + .root_turn_id() + .unwrap_or_else(|| turn_context.sub_id.clone()), + ), + trace_id: turn_context.trace_id.clone(), + started_at: turn_context.turn_timing_state.started_at_unix_secs().await, + model_context_window: turn_context.model_context_window(), + collaboration_mode_kind: turn_context.mode(), + }; + self.send_event(turn_context, EventMsg::TurnStarted(event)) + .await; + } + /// Persist the event to rollout and send it to clients. pub(crate) async fn send_event(&self, turn_context: &TurnContext, msg: EventMsg) { let legacy_source = msg.clone(); diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index cd75eaac8e..f85604e77b 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -94,6 +94,7 @@ fn completed_user_turn_rollout( RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -351,6 +352,7 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -423,6 +425,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -459,6 +462,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: rolled_back_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -543,6 +547,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -576,6 +581,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: incomplete_turn_id, + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -642,6 +648,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -675,6 +682,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: second_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -707,6 +715,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: standalone_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -779,6 +788,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -812,6 +822,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: assistant_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -876,6 +887,7 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: only_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -933,6 +945,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: user_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -965,6 +978,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: standalone_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1013,6 +1027,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: previous_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1044,6 +1059,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: incomplete_turn_id, + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1555,6 +1571,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: current_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1632,6 +1649,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: previous_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1768,6 +1786,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: previous_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1799,6 +1818,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: aborted_turn_id, + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1902,6 +1922,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: previous_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -1933,6 +1954,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: current_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -2038,6 +2060,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: previous_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -2069,6 +2092,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: incomplete_turn_id, + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -2132,6 +2156,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_preserves_turn_ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: current_turn_id, + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -2218,6 +2243,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: previous_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -2249,6 +2275,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: compacted_incomplete_turn_id, + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -2283,6 +2310,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: replacing_turn_id, + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index aa828752a3..169b055be5 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -618,6 +618,7 @@ async fn regular_turn_emits_turn_started_with_trace_id_without_waiting_for_start }; assert_eq!(turn_started.turn_id, tc.sub_id); assert_eq!(turn_started.trace_id, tc.trace_id); + assert_eq!(turn_started.root_turn_id, Some(tc.sub_id.clone())); sess.abort_all_tasks(TurnAbortReason::Interrupted).await; } @@ -3839,6 +3840,7 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4046,6 +4048,7 @@ async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4077,6 +4080,7 @@ async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: rolled_back_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4170,6 +4174,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: first_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4199,6 +4204,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: compact_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4236,6 +4242,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: rolled_back_turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4319,6 +4326,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4348,6 +4356,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-2".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), @@ -4377,6 +4386,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() { RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-3".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index ba4b44a8b5..1dc274437f 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -10,8 +10,6 @@ use crate::session::turn::run_turn; use crate::session::turn_context::TurnContext; use crate::session_startup_prewarm::SessionStartupPrewarmResolution; use crate::state::TaskKind; -use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::TurnStartedEvent; use codex_thread_store::PersistContext; use tracing::Instrument; use tracing::trace_span; @@ -48,14 +46,7 @@ impl SessionTask for RegularTask { // Regular turns emit `TurnStarted` inline so first-turn lifecycle does // not wait on startup prewarm resolution. let prewarmed_client_session = async { - let event = EventMsg::TurnStarted(TurnStartedEvent { - turn_id: ctx.sub_id.clone(), - trace_id: ctx.trace_id.clone(), - started_at: ctx.turn_timing_state.started_at_unix_secs().await, - model_context_window: ctx.model_context_window(), - collaboration_mode_kind: ctx.mode(), - }); - sess.send_event(ctx.as_ref(), event).await; + sess.emit_turn_started(&ctx).await; sess.set_server_reasoning_included(/*included*/ false).await; sess.consume_startup_prewarm_for_regular_turn(&cancellation_token) .await diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index 97f43df05a..91ed68a51b 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -36,7 +36,6 @@ use codex_protocol::items::TurnItem; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecCommandSource; -use codex_protocol::protocol::TurnStartedEvent; use codex_sandboxing::SandboxType; use codex_shell_command::parse_command::parse_command; use codex_thread_store::PersistContext; @@ -123,14 +122,7 @@ pub(crate) async fn execute_user_shell_command( // standalone lifecycle tasks (for example /shell, and review once it emits TurnStarted). // `/compact` is an intentional exception because compaction requests should not include // freshly reinjected context before the summary/replacement history is applied. - let event = EventMsg::TurnStarted(TurnStartedEvent { - turn_id: turn_context.sub_id.clone(), - trace_id: turn_context.trace_id.clone(), - started_at: turn_context.turn_timing_state.started_at_unix_secs().await, - model_context_window: turn_context.model_context_window(), - collaboration_mode_kind: turn_context.mode(), - }); - session.send_event(turn_context.as_ref(), event).await; + session.emit_turn_started(&turn_context).await; } let Some((turn_environment, environment_shell)) = turn_context diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 7103c17145..c25a55575c 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -562,6 +562,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() { RolloutItem::ResponseItem(assistant_msg("a1").into()), RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-2".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2705,6 +2706,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() { InitialHistory::Forked(vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-explicit".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/core/src/thread_rollout_truncation_tests.rs b/codex-rs/core/src/thread_rollout_truncation_tests.rs index 88d496eaa1..71a80665e2 100644 --- a/codex-rs/core/src/thread_rollout_truncation_tests.rs +++ b/codex-rs/core/src/thread_rollout_truncation_tests.rs @@ -79,6 +79,7 @@ fn inter_agent_communication(text: &str, trigger_turn: bool) -> RolloutItem { fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/core/tests/suite/agent_execution.rs b/codex-rs/core/tests/suite/agent_execution.rs index ace2f50a74..b31d821a49 100644 --- a/codex-rs/core/tests/suite/agent_execution.rs +++ b/codex-rs/core/tests/suite/agent_execution.rs @@ -205,6 +205,62 @@ async fn v2_nested_spawn_checks_shared_active_execution_capacity() -> Result<()> Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn child_turn_start_preserves_root_attribution() -> Result<()> { + let server = start_mock_server().await; + mount_root_collaboration_call( + &server, + FIRST_PROMPT, + "first-call", + "spawn_agent", + json!({ + "message": FIRST_TASK, "task_name": "first", "fork_turns": "none", + }), + ) + .await; + let worker = mount_completed_worker(&server, FIRST_TASK, "first-call").await; + let test = test_codex() + .with_model("gpt-5.6-sol") + .with_config(|config| { + config.features.enable(Feature::Collab).unwrap(); + config.features.enable(Feature::MultiAgentV2).unwrap(); + }) + .build_with_auto_env(&server) + .await?; + test.submit_turn(FIRST_PROMPT).await?; + let thread_ids = test.thread_manager.list_thread_ids().await; + assert_eq!(thread_ids.len(), 2); + let mut starts = Vec::new(); + for thread_id in thread_ids { + let thread = test.thread_manager.get_thread(thread_id).await?; + if thread_id != test.session_configured.thread_id { + wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await; + } + thread.flush_rollout().await?; + let history = thread.load_history(/*include_archived*/ false).await?; + for item in history.items { + if let codex_history::RolloutItem::EventMsg(EventMsg::TurnStarted(event)) = item { + starts.push((thread_id, event)); + } + } + } + worker.single_request(); + assert_eq!(starts.len(), 2); + assert_ne!(starts[0].1.turn_id, starts[1].1.turn_id); + let root_turn_id = &starts + .iter() + .find(|(id, _)| *id == test.session_configured.thread_id) + .expect("root turn") + .1 + .turn_id; + assert!( + starts + .iter() + .all(|(_, event)| { event.root_turn_id.as_ref() == Some(root_turn_id) }) + ); + Ok(()) +} + #[derive(Clone, Copy, PartialEq, Eq)] enum ResidencyReload { Sender, diff --git a/codex-rs/core/tests/suite/resume_warning.rs b/codex-rs/core/tests/suite/resume_warning.rs index df73438792..3e4f6c199e 100644 --- a/codex-rs/core/tests/suite/resume_warning.rs +++ b/codex-rs/core/tests/suite/resume_warning.rs @@ -62,6 +62,7 @@ fn resume_history( history: Arc::new(vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/core/tests/suite/tools.rs b/codex-rs/core/tests/suite/tools.rs index 5ff41881cf..5afee5f17b 100644 --- a/codex-rs/core/tests/suite/tools.rs +++ b/codex-rs/core/tests/suite/tools.rs @@ -45,6 +45,7 @@ use core_test_support::submit_thread_settings; use core_test_support::test_codex::local; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use test_case::test_case; @@ -123,20 +124,25 @@ async fn strict_tool_collisions_fail_the_turn_before_sampling( defer_loading: false, })] }; - let thread = test + let codex_core::NewThread { thread, .. } = test .thread_manager .start_thread(StartThreadOptions { dynamic_tools, ..StartThreadOptions::new(test.config.clone()) }) - .await? - .thread; + .await?; thread - .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { - text: "use the planning tool".to_string(), - text_elements: Vec::new(), - }])) + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "use the planning tool".to_string(), + text_elements: Vec::new(), + }]) + .on_start(codex_core::TurnStartOptions { + root_turn_id: Some("root-turn".into()), + ..Default::default() + }), + ) .await?; let EventMsg::Error(error) = @@ -157,6 +163,18 @@ async fn strict_tool_collisions_fail_the_turn_before_sampling( unreachable!("event predicate guarantees turn completion"); }; assert_eq!(completed.error, Some(error)); + thread.flush_rollout().await?; + let history = thread.load_history(/*include_archived*/ false).await?; + let attribution = history.items.iter().find_map(|item| match item { + codex_history::RolloutItem::EventMsg(EventMsg::TurnStarted(event)) + if event.turn_id == completed.turn_id => + { + event.root_turn_id.as_deref() + } + _ => None, + }); + assert_eq!(attribution, Some("root-turn")); + assert!( server .received_requests() diff --git a/codex-rs/external-agent-migration/src/sessions/export.rs b/codex-rs/external-agent-migration/src/sessions/export.rs index 2370bf1fc1..e3b360da3a 100644 --- a/codex-rs/external-agent-migration/src/sessions/export.rs +++ b/codex-rs/external-agent-migration/src/sessions/export.rs @@ -106,6 +106,7 @@ pub(super) fn rollout_items_from_messages(messages: Vec) -> items.push(RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { turn_id: turn_id.clone(), + root_turn_id: None, trace_id: None, started_at, model_context_window: None, diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index e25cce88a0..0c79853a35 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2174,6 +2174,10 @@ pub struct TurnCompleteEvent { #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct TurnStartedEvent { pub turn_id: String, + /// ID of the originating turn in the root thread; equals `turn_id` for root turns. + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub root_turn_id: Option, // Persist for rollout consumers that correlate turns with telemetry traces. #[serde(default, skip_serializing_if = "Option::is_none")] #[ts(optional)] @@ -4468,6 +4472,16 @@ mod tests { use tempfile::NamedTempFile; use tempfile::TempDir; + #[test] + fn old_turn_started_records_have_no_root_attribution() { + let event: TurnStartedEvent = serde_json::from_value(serde_json::json!({ + "turn_id": "old-turn", + "model_context_window": null + })) + .unwrap(); + assert_eq!(event.root_turn_id, None); + } + #[test] fn review_decision_denied_round_trip() -> Result<()> { let decision = ReviewDecision::Denied { diff --git a/codex-rs/rollout/src/persistence_metrics_tests.rs b/codex-rs/rollout/src/persistence_metrics_tests.rs index fed1b112e8..264ca889c6 100644 --- a/codex-rs/rollout/src/persistence_metrics_tests.rs +++ b/codex-rs/rollout/src/persistence_metrics_tests.rs @@ -42,6 +42,7 @@ fn retained_message(text: &str) -> RolloutItem { fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 37467db3e1..4e20031a50 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -1076,6 +1076,7 @@ mod tests { .append_items(&[RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, 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 55cae4f6d2..0d7d90a892 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -776,6 +776,7 @@ fn append_items(path: &Path, items: impl IntoIterator) { fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: Some(128_000), diff --git a/codex-rs/thread-store/src/local/revert_thread_tests.rs b/codex-rs/thread-store/src/local/revert_thread_tests.rs index 137dbaeeee..0b27ac1b82 100644 --- a/codex-rs/thread-store/src/local/revert_thread_tests.rs +++ b/codex-rs/thread-store/src/local/revert_thread_tests.rs @@ -279,6 +279,7 @@ async fn turn_ids(store: &LocalThreadStore, thread_id: ThreadId) -> Vec fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, diff --git a/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs b/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs index 13a80e37f8..8c446ccc5c 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs @@ -348,6 +348,7 @@ impl LegacyRolloutCanonicalizer { timestamp, RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id, + root_turn_id: None, trace_id: None, started_at: Some(started_at), model_context_window: None, diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 473ead2502..8dfad90b9e 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -223,6 +223,7 @@ fn item_completed(turn_id: &str, item_id: &str) -> RolloutItem { fn started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: Some(1_735_905_600), model_context_window: None, 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 2ea0ac86ef..3f6fcbf615 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 @@ -2601,6 +2601,7 @@ async fn create_paginated_subagent_thread( fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: Some(10), model_context_window: None, diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 791c0ff372..fb7d4d8c78 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -631,6 +631,7 @@ mod tests { .observe_appended_items(&[RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { turn_id: "turn-1".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 1cb3cf299f..3397652530 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -7982,6 +7982,7 @@ async fn prompt_edit_forks_before_selected_prompt_and_preserves_source() -> Resu for item in [ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, diff --git a/codex-rs/tui/src/app/tests/pagination_completion_tests.rs b/codex-rs/tui/src/app/tests/pagination_completion_tests.rs index b0418b8324..54bd23ba1f 100644 --- a/codex-rs/tui/src/app/tests/pagination_completion_tests.rs +++ b/codex-rs/tui/src/app/tests/pagination_completion_tests.rs @@ -53,6 +53,7 @@ async fn older_pagination_completion_footers_follow_answers_without_overlap_dupl let finished = completed_at.timestamp() + index as i64 * 60; let mut events = vec![EventMsg::TurnStarted(TurnStartedEvent { turn_id: turn_id.clone(), + root_turn_id: None, trace_id: None, started_at: Some(finished - 125), model_context_window: None, diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index f8d053236d..2b3fca5cb3 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -1824,6 +1824,7 @@ async fn older_pagination_reconciles_review_prompts_across_page_boundaries() -> ]); let events = std::iter::once(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "cross-page-review-turn".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -1986,6 +1987,7 @@ async fn transcript_home_loads_every_older_history_page() -> Result<()> { .collect::, _>>()?; let events = std::iter::once(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "multi-page-turn".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, @@ -2394,6 +2396,7 @@ async fn underfilled_scrollback_fetches_older_pages_without_opening_the_transcri .collect::, _>>()?; let events = std::iter::once(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "scrollback-pagination-turn".to_string(), + root_turn_id: None, trace_id: None, started_at: None, model_context_window: None,