Preserve root turn attribution in turn-start events (#44611)

## Why

Persisted turn-start events lack the originating root turn ID, so they cannot directly associate child turns with the root turn that initiated them.

## What changed

- Add optional `root_turn_id` to `TurnStartedEvent`. Use the inherited root turn ID when available, otherwise the turn's own ID.
- Centralize turn-start emission for regular turns, shell commands, and manual compaction so they record attribution consistently.
- Carry attribution into `ThreadHistoryTurnChange` when processing turn-start events and retain it through completion in the history builder.
- Keep older records compatible by defaulting missing attribution to `None`.

## Testing

Add coverage for root and child turn attribution, persistence when tool collisions fail a turn before sampling, history change propagation, and deserialization of older turn-start records.

GitOrigin-RevId: 54a80b828ca7d1c4c19f13d468c01d052e1b9130

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,

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,

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()