Preserve turn triggers across delegated agent work (#44659)

## Why

Delegated usage should retain the trigger of the turn that initiated it, including composer input and scheduled automation.

## What changed

Propagate `turn_trigger` through agent spawning and follow-up messages in both multi-agent tool implementations. Apply the supplied trigger when pending mailbox work starts a new turn.

## Testing

Extend tests to verify trigger propagation to spawned and nested agents, updated triggers on reused agents and peer follow-ups, and preservation of the active turn's trigger when steering.

GitOrigin-RevId: e8fea208587c633fec7cb3f1c0815ff91d1a44e5

codex-rs/core/src/agent/control.rs

@@ -93,6 +93,8 @@ pub(crate) struct SpawnAgentOptions {
     pub(crate) fork_mode: Option<SpawnAgentForkMode>,
     pub(crate) parent_thread_id: Option<ThreadId>,
     pub(crate) parent_turn_id: Option<String>,
+    /// Attribute delegated usage to the turn that initiated it.
+    pub(crate) turn_trigger: Option<String>,
     pub(crate) root_turn_id: Option<String>,
     pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
     pub(crate) multi_agent_v2_usage_hints: Option<ResolvedMultiAgentV2UsageHints>,

codex-rs/core/src/agent/control/spawn.rs

@@ -765,6 +765,7 @@ impl AgentControl {
 
         let start_options = TurnStartOptions {
             parent_turn_id: options.parent_turn_id,
+            turn_trigger: options.turn_trigger,
             root_turn_id: options.root_turn_id,
             cyber_access_program: options.cyber_access_program,
             ..Default::default()

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

@@ -858,6 +858,9 @@ async fn steer_only_requires_active_turn() {
 #[tokio::test]
 async fn steer_only_enforces_expected_turn_id() {
     let (session, turn_context, _rx) = make_session_and_context_with_rx().await;
+    turn_context
+        .turn_metadata_state
+        .set_turn_trigger("composer".to_string());
     session
         .spawn_task(
             Arc::clone(&turn_context),
@@ -904,7 +907,12 @@ async fn steer_only_enforces_expected_turn_id() {
 
     let submission = handle(
         &session,
-        TurnInputRequest::new(SubmittedTurnInput::ResponseItem(output)),
+        TurnInputRequest::new(SubmittedTurnInput::ResponseItem(output)).on_start(
+            TurnStartOptions {
+                turn_trigger: Some("automation_cron_scheduled".to_string()),
+                ..Default::default()
+            },
+        ),
         TurnInputMode::StartOrSteer,
         "test-submission".to_string(),
     )
@@ -917,6 +925,13 @@ async fn steer_only_enforces_expected_turn_id() {
             turn_id: turn_context.sub_id.clone()
         }
     );
+    assert_eq!(
+        turn_context
+            .turn_metadata_state
+            .current_turn_trigger()
+            .as_deref(),
+        Some("composer")
+    );
     let turn_state = session
         .input_queue
         .turn_state_for_sub_id(&session.active_turn, &turn_context.sub_id)

codex-rs/core/src/tasks/mod.rs

@@ -478,6 +478,9 @@ impl Session {
                 },
             )
             .await;
+        if let Some(trigger) = start_options.turn_trigger {
+            turn_context.turn_metadata_state.set_turn_trigger(trigger);
+        }
         if let Some(id) = start_options.parent_turn_id {
             if let Some(initiating_agent_path) = input.iter().find_map(|item| {
                 let TurnInput::InterAgentCommunication(communication) = item else {

codex-rs/core/src/tools/handlers/multi_agents/send_input.rs

@@ -93,6 +93,7 @@ impl Handler {
                 crate::TurnStartOptions {
                     parent_turn_id: Some(turn.sub_id.clone()),
                     root_turn_id: turn.turn_metadata_state.root_turn_id(),
+                    turn_trigger: turn.turn_metadata_state.current_turn_trigger(),
                     cyber_access_program: turn.cyber_access_program,
                     ..Default::default()
                 },

codex-rs/core/src/tools/handlers/multi_agents/spawn.rs

@@ -124,6 +124,7 @@ async fn handle_spawn_agent(
             parent_thread_id: Some(session.thread_id),
             parent_turn_id: Some(turn.sub_id.clone()),
             root_turn_id: turn.turn_metadata_state.root_turn_id(),
+            turn_trigger: turn.turn_metadata_state.current_turn_trigger(),
             environments: Some(step_context.environments.to_selections()),
             multi_agent_v2_usage_hints: None,
             cyber_access_program: turn.cyber_access_program,

codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs

@@ -120,6 +120,7 @@ pub(super) async fn handle_message_string_tool(
             crate::TurnStartOptions {
                 parent_turn_id,
                 root_turn_id: turn.turn_metadata_state.root_turn_id(),
+                turn_trigger: turn.turn_metadata_state.current_turn_trigger(),
                 cyber_access_program: turn.cyber_access_program,
                 ..Default::default()
             },

codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs

@@ -219,6 +219,7 @@ async fn handle_spawn_agent(
                     parent_thread_id: Some(session.thread_id),
                     parent_turn_id: Some(turn.sub_id.clone()),
                     root_turn_id: turn.turn_metadata_state.root_turn_id(),
+                    turn_trigger: turn.turn_metadata_state.current_turn_trigger(),
                     environments: Some(step_context.environments.to_selections()),
                     multi_agent_v2_usage_hints,
                     cyber_access_program: turn.cyber_access_program,

codex-rs/core/src/turn_metadata.rs

@@ -340,6 +340,10 @@ impl TurnMetadataState {
         let _ = self.turn_trigger.set(turn_trigger);
     }
 
+    pub(crate) fn current_turn_trigger(&self) -> Option<String> {
+        self.turn_trigger.get().cloned()
+    }
+
     pub(crate) fn root_turn_id(&self) -> Option<String> {
         self.root_turn_id.get().cloned()
     }

codex-rs/core/tests/suite/subagent_notifications.rs

@@ -2,6 +2,7 @@ use anyhow::Result;
 use codex_core::StartThreadOptions;
 use codex_core::ThreadConfigSnapshot;
 use codex_core::TurnInputRequest;
+use codex_core::TurnStartOptions;
 use codex_core::config::AgentRoleConfig;
 use codex_core::config::CurrentTimeReminderConfig;
 use codex_features::Feature;
@@ -1058,7 +1059,7 @@ async fn spawned_child_receives_forked_parent_context(
     test.submit_turn(TURN_0_FORK_PROMPT).await?;
     let _ = seed_turn.single_request();
 
-    test.submit_turn(TURN_1_PROMPT).await?;
+    submit_turn_with_trigger(&test, TURN_1_PROMPT, "composer").await?;
     let parent_body = spawn_turn.single_request().body_json();
 
     let child_request = wait_for_request_with_model(&child_request_log, REQUESTED_MODEL).await?;
@@ -1069,7 +1070,13 @@ async fn spawned_child_receives_forked_parent_context(
             .as_str()
             .expect("child turn metadata"),
     )?;
-    assert_eq!(child_metadata["thread_source"], "subagent");
+    assert_eq!(
+        (
+            &child_metadata["thread_source"],
+            &child_metadata["turn_trigger"]
+        ),
+        (&json!("subagent"), &json!("composer")),
+    );
     let original_parent_turn_id = parent_body["client_metadata"]["turn_id"]
         .as_str()
         .expect("legacy spawn parent turn id");
@@ -1127,7 +1134,7 @@ async fn spawned_child_receives_forked_parent_context(
     )
     .await;
 
-    test.submit_turn("reuse the legacy child").await?;
+    submit_turn_with_trigger(&test, "reuse the legacy child", "automation_cron_scheduled").await?;
     let followup_parent_body = parent.single_request().body_json();
     let reused_child_body = wait_for_request_with_model(&followup, REQUESTED_MODEL)
         .await?
@@ -1142,6 +1149,32 @@ async fn spawned_child_receives_forked_parent_context(
     assert_parent_turn(&reused_child_body, Some(followup_parent_turn_id))?;
     assert_root_turn(&followup_parent_body, Some(followup_parent_turn_id))?;
     assert_root_turn(&reused_child_body, Some(followup_parent_turn_id))?;
+    let reused_metadata: Value = serde_json::from_str(
+        reused_child_body["client_metadata"]["x-codex-turn-metadata"]
+            .as_str()
+            .expect("reused child turn metadata"),
+    )?;
+    assert_eq!(reused_metadata["turn_trigger"], "automation_cron_scheduled");
+    Ok(())
+}
+
+async fn submit_turn_with_trigger(test: &TestCodex, prompt: &str, trigger: &str) -> Result<()> {
+    test.codex
+        .start_or_steer_turn(
+            TurnInputRequest::user_input(vec![UserInput::Text {
+                text: prompt.to_string(),
+                text_elements: Vec::new(),
+            }])
+            .on_start(TurnStartOptions {
+                turn_trigger: Some(trigger.to_string()),
+                ..Default::default()
+            }),
+        )
+        .await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
     Ok(())
 }
 
@@ -1322,7 +1355,7 @@ async fn grandchild_full_fork_preserves_context_baseline(
         .build_with_auto_env(&server)
         .await?;
 
-    test.submit_turn(ROOT_PROMPT).await?;
+    submit_turn_with_trigger(&test, ROOT_PROMPT, "automation_heartbeat_scheduled").await?;
     let root_request = root_log.single_request();
     let mut descendant_requests = Vec::new();
     for (mock, agent_name) in [
@@ -1363,6 +1396,15 @@ async fn grandchild_full_fork_preserves_context_baseline(
                 AgentStatus::Completed(Some("done".to_string()))
             );
         }
+        let metadata: Value = serde_json::from_str(
+            request.body_json()["client_metadata"]["x-codex-turn-metadata"]
+                .as_str()
+                .expect("descendant turn metadata"),
+        )?;
+        assert_eq!(
+            (&metadata["thread_source"], &metadata["turn_trigger"]),
+            (&json!("subagent"), &json!("automation_heartbeat_scheduled")),
+        );
         descendant_requests.push(request);
     }
     let context_counts = [
@@ -2746,7 +2788,7 @@ async fn multi_agent_v2_peer_followup_completion_notifies_initiating_turn() -> R
     )
     .await;
 
-    test.submit_turn(SPAWN_WORKER_PROMPT).await?;
+    submit_turn_with_trigger(&test, SPAWN_WORKER_PROMPT, "automation_cron_scheduled").await?;
     let worker_thread_id = created_threads.recv().await?;
     let worker_thread = test.thread_manager.get_thread(worker_thread_id).await?;
     wait_for_event(worker_thread.as_ref(), |event| {
@@ -2845,7 +2887,7 @@ async fn multi_agent_v2_peer_followup_completion_notifies_initiating_turn() -> R
         );
     }
 
-    test.submit_turn(SPAWN_REQUESTER_PROMPT).await?;
+    submit_turn_with_trigger(&test, SPAWN_REQUESTER_PROMPT, "composer").await?;
     let requester_thread_id = created_threads.recv().await?;
     let requester_thread = test.thread_manager.get_thread(requester_thread_id).await?;
     let requester_turn_id = wait_for_event_match(requester_thread.as_ref(), |event| match event {
@@ -2874,6 +2916,13 @@ async fn multi_agent_v2_peer_followup_completion_notifies_initiating_turn() -> R
             (body["client_metadata"]["thread_id"] == json!(worker_thread_id)
                 && request.body_contains_text(WORKER_FOLLOWUP_TASK))
             .then(|| {
+                let metadata: Value = serde_json::from_str(
+                    body["client_metadata"]["x-codex-turn-metadata"]
+                        .as_str()
+                        .expect("worker turn metadata"),
+                )
+                .expect("worker turn metadata JSON");
+                assert_eq!(metadata["turn_trigger"], "composer");
                 body["client_metadata"]["turn_id"]
                     .as_str()
                     .expect("worker follow-up turn ID")