Establish root turn identity for independent tasks and memory requests (#42900)

## Why

Background and empty-input turns could lack a `root_turn_id`, and detached memory requests omitted turn identity entirely. Tasks without a root could also adopt one from coalesced mailbox input.

## What changed

- Set `root_turn_id` to the task's turn ID at task start unless an inherited or recovered root is already present.
- Stop assigning roots from mailbox input or steering an active turn.
- Give detached memory requests a fresh UUID for both `turn_id` and `root_turn_id`, included in turn headers and client metadata while still omitting session and thread identity.

## Testing

Extend assertions for background turns, empty input, goal continuations, manual compaction, and detached memory requests. Exercise mailbox coalescing with both independent and inherited roots to verify the task keeps its established root.

GitOrigin-RevId: ee0f2f1b43542a9f5884f694c1d94341b476779c

codex-rs/core/src/session/input_queue.rs

@@ -292,14 +292,10 @@ impl InputQueue {
         &self,
         active_turn: &Mutex<Option<ActiveTurn>>,
     ) -> (Vec<TurnInput>, TurnStartOptions) {
-        let (pending_input, accepts_mailbox_delivery, active_turn_metadata) = {
+        let (pending_input, accepts_mailbox_delivery) = {
             let mut active = active_turn.lock().await;
             match active.as_mut() {
                 Some(active_turn) => {
-                    let active_turn_metadata = active_turn
-                        .task
-                        .as_ref()
-                        .map(|task| Arc::clone(&task.turn_context.turn_metadata_state));
                     let mut turn_state = active_turn.turn_state.lock().await;
                     let accepts_mailbox_delivery =
                         turn_state.accepts_mailbox_delivery_for_current_turn();
@@ -308,25 +304,15 @@ impl InputQueue {
                     } else {
                         Vec::new()
                     };
-                    (
-                        pending_input,
-                        accepts_mailbox_delivery,
-                        active_turn_metadata,
-                    )
+                    (pending_input, accepts_mailbox_delivery)
                 }
-                None => (Vec::new(), true, None),
+                None => (Vec::new(), true),
             }
         };
         if !accepts_mailbox_delivery {
             return (pending_input, TurnStartOptions::default());
         }
         let (mailbox_items, start_options) = self.drain_mailbox_input_items().await;
-        if let Some(active_turn_metadata) = active_turn_metadata
-            && active_turn_metadata.root_turn_id().is_none()
-            && let Some(root_turn_id) = start_options.root_turn_id.as_ref()
-        {
-            active_turn_metadata.set_root_turn_id(root_turn_id.clone());
-        }
         if pending_input.is_empty() {
             (mailbox_items, start_options)
         } else {

codex-rs/core/src/session/review.rs

@@ -130,9 +130,6 @@ pub(super) async fn spawn_review_thread(
         auto_review_enabled,
         &model_info,
     ));
-    if turn_metadata_state.can_start_root_turn(&session_source) {
-        turn_metadata_state.set_root_turn_id(review_turn_id.clone());
-    }
 
     let extension_data = Arc::new(codex_extension_api::ExtensionData::new(
         review_turn_id.clone(),

codex-rs/core/src/session/tests.rs

@@ -12053,20 +12053,14 @@ async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
     assert!(sess.input_queue.has_trigger_turn_mailbox_items().await);
 }
 
+#[test_case(None; "independent root")]
+#[test_case(Some("root-a"); "inherited root")]
 #[tokio::test]
-async fn active_turn_keeps_first_root_when_mail_coalesces() {
+async fn active_turn_keeps_first_root_when_mail_coalesces(inherited_root: Option<&str>) {
     let (sess, tc, _rx) = make_session_and_context_with_rx().await;
-    tc.turn_metadata_state
-        .set_root_turn_id("root-a".to_string());
-    sess.spawn_task(
-        Arc::clone(&tc),
-        Vec::new(),
-        NeverEndingTask {
-            kind: TaskKind::Regular,
-            listen_to_cancellation_token: true,
-        },
-    )
-    .await;
+    if let Some(root) = inherited_root {
+        tc.turn_metadata_state.set_root_turn_id(root.to_string());
+    }
     let first = InterAgentCommunication::new(
         AgentPath::try_from("/root/worker_a").expect("worker path should parse"),
         AgentPath::root(),
@@ -12081,10 +12075,13 @@ async fn active_turn_keeps_first_root_when_mail_coalesces() {
         "second".to_string(),
         /*trigger_turn*/ true,
     );
-    for (communication, parent_turn_id, root_turn_id) in [
+    for (index, (communication, parent_turn_id, root_turn_id)) in [
         (first.clone(), "parent-a", "root-a"),
         (second.clone(), "parent-b", "root-b"),
-    ] {
+    ]
+    .into_iter()
+    .enumerate()
+    {
         sess.input_queue
             .enqueue_mailbox_communication(
                 communication,
@@ -12095,6 +12092,19 @@ async fn active_turn_keeps_first_root_when_mail_coalesces() {
                 },
             )
             .await;
+        if index == 0 {
+            // The first message is already queued when this independent task
+            // starts; the second arrives after its root is established.
+            sess.spawn_task(
+                Arc::clone(&tc),
+                Vec::new(),
+                NeverEndingTask {
+                    kind: TaskKind::Regular,
+                    listen_to_cancellation_token: true,
+                },
+            )
+            .await;
+        }
     }
 
     assert_eq!(
@@ -12106,7 +12116,7 @@ async fn active_turn_keeps_first_root_when_mail_coalesces() {
     );
     assert_eq!(
         tc.turn_metadata_state.root_turn_id().as_deref(),
-        Some("root-a")
+        Some(inherited_root.unwrap_or(&tc.sub_id))
     );
     assert!(!sess.input_queue.has_pending_mailbox_items().await);
 

codex-rs/core/src/session/turn_input.rs

@@ -262,11 +262,6 @@ async fn start_or_steer(
             ));
         }
     };
-    let can_start_root_turn = start.parent_turn_id.is_none() && start.root_turn_id.is_none();
-    let incoming_root_turn_id = start
-        .parent_turn_id
-        .as_ref()
-        .map(|_| start.root_turn_id.clone());
     let settings = PreparedTurnInputSettings::prepare(session, thread_settings, start).await?;
     match session
         .steer_input(
@@ -275,7 +270,6 @@ async fn start_or_steer(
             /*expected_turn_id*/ None,
             settings.required_active_final_output_json_schema(),
             responsesapi_client_metadata.clone(),
-            incoming_root_turn_id,
         )
         .await
     {
@@ -290,16 +284,6 @@ async fn start_or_steer(
             else {
                 unreachable!("explicit user input can enter Plan mode");
             };
-            if can_start_root_turn
-                && has_explicit_input
-                && turn_context
-                    .turn_metadata_state
-                    .can_start_root_turn(&turn_context.session_source)
-            {
-                turn_context
-                    .turn_metadata_state
-                    .set_root_turn_id(submission_id.clone());
-            }
             if let Some(responsesapi_client_metadata) = responsesapi_client_metadata {
                 turn_context
                     .turn_metadata_state
@@ -340,7 +324,6 @@ async fn start_if_idle(
         responsesapi_client_metadata,
         ..
     } = request;
-    let can_start_root_turn = start.parent_turn_id.is_none() && start.root_turn_id.is_none();
     if session.input_queue.has_trigger_turn_mailbox_items().await {
         return Ok(TurnInputSubmission::NotSubmitted {
             reason: NotSubmittedReason::PendingTriggerTurn,
@@ -403,16 +386,6 @@ async fn start_if_idle(
             .turn_metadata_state
             .set_responsesapi_client_metadata(responsesapi_client_metadata);
     }
-    if kind == TurnStartKind::User
-        && can_start_root_turn
-        && turn_context
-            .turn_metadata_state
-            .can_start_root_turn(&turn_context.session_source)
-    {
-        turn_context
-            .turn_metadata_state
-            .set_root_turn_id(submission_id.clone());
-    }
     session
         .maybe_emit_model_warnings_for_turn(turn_context.as_ref())
         .await;
@@ -469,10 +442,6 @@ async fn steer(
             "only user input can steer a turn".to_string(),
         ));
     }
-    let incoming_root_turn_id = start
-        .parent_turn_id
-        .as_ref()
-        .map(|_| start.root_turn_id.clone());
     let settings = PreparedTurnInputSettings::prepare(session, thread_settings, start).await?;
     match session
         .steer_input(
@@ -481,7 +450,6 @@ async fn steer(
             Some(expected_turn_id.as_str()),
             settings.required_active_final_output_json_schema(),
             responsesapi_client_metadata,
-            incoming_root_turn_id,
         )
         .await
     {
@@ -557,7 +525,6 @@ impl Session {
         expected_turn_id: Option<&str>,
         required_final_output_json_schema: Option<&Value>,
         responsesapi_client_metadata: Option<HashMap<String, String>>,
-        incoming_root_turn_id: Option<Option<String>>,
     ) -> Result<String, NotSubmittedReason> {
         let mut active = self.active_turn.lock().await;
         let Some(active_turn) = active.as_mut() else {
@@ -627,18 +594,6 @@ impl Session {
             input => pending_turn_input(self, input.clone()).await,
         };
         pending_input.push(input);
-        if active_task
-            .turn_context
-            .turn_metadata_state
-            .root_turn_id()
-            .is_none()
-            && let Some(Some(incoming_root_turn_id)) = incoming_root_turn_id
-        {
-            active_task
-                .turn_context
-                .turn_metadata_state
-                .set_root_turn_id(incoming_root_turn_id);
-        }
         self.input_queue
             .extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
                 active_turn.turn_state.as_ref(),