fix(core): preserve spawned-agent task payload and profile

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

@@ -60,6 +60,7 @@ pub(crate) enum SpawnAgentForkMode {
 pub(crate) struct SpawnAgentOptions {
     pub(crate) fork_parent_spawn_call_id: Option<String>,
     pub(crate) fork_mode: Option<SpawnAgentForkMode>,
+    pub(crate) initial_task_message_id: Option<String>,
     pub(crate) parent_thread_id: Option<ThreadId>,
     pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
 }
@@ -175,6 +176,34 @@ impl AgentControl {
         result
     }
 
+    pub(crate) async fn send_initial_agent_task(
+        &self,
+        agent_id: ThreadId,
+        message_id: String,
+        communication: InterAgentCommunication,
+    ) -> CodexResult<String> {
+        let last_task_message = last_task_message_from_communication(&communication);
+        let state = self.upgrade()?;
+        let result = self
+            .handle_thread_request_result(
+                agent_id,
+                &state,
+                state
+                    .deliver_initial_agent_task(agent_id, message_id, communication)
+                    .await,
+            )
+            .await;
+        if result.is_ok() {
+            match last_task_message {
+                Some(last_task_message) => self
+                    .state
+                    .update_last_task_message(agent_id, last_task_message),
+                None => self.state.clear_last_task_message(agent_id),
+            }
+        }
+        result
+    }
+
     /// Returns the wire protocol of `agent_id`'s configured provider, or `None`
     /// when the thread is not currently resolvable. Callers use this to encode an
     /// inter-agent message in a form the receiver's wire can actually deliver.

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

@@ -346,8 +346,20 @@ impl AgentControl {
         )
         .await;
 
-        self.send_input(new_thread.thread_id, initial_operation)
-            .await?;
+        match (options.initial_task_message_id, initial_operation) {
+            (Some(initial_task_message_id), Op::InterAgentCommunication { communication }) => {
+                self.send_initial_agent_task(
+                    new_thread.thread_id,
+                    initial_task_message_id,
+                    communication,
+                )
+                .await?;
+            }
+            (_, initial_operation) => {
+                self.send_input(new_thread.thread_id, initial_operation)
+                    .await?;
+            }
+        }
         if multi_agent_version != MultiAgentVersion::V2 {
             let child_reference = agent_metadata
                 .agent_path

codex-rs/core/src/codex_thread.rs

@@ -273,6 +273,20 @@ impl CodexThread {
         Ok(())
     }
 
+    /// Queues a spawned agent's initial task with a stable id for exact-once model delivery.
+    pub async fn enqueue_initial_agent_task_with_id(
+        &self,
+        message_id: String,
+        communication: InterAgentCommunication,
+    ) -> CodexResult<()> {
+        if !self.is_running() {
+            return Err(CodexErr::InternalAgentDied);
+        }
+        crate::session::enqueue_initial_agent_task(&self.codex.session, message_id, communication)
+            .await?;
+        Ok(())
+    }
+
     pub async fn queued_mailbox_messages(&self) -> Vec<QueuedMailboxMessage> {
         self.codex.session.input_queue.list_mailbox_messages().await
     }

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

@@ -463,6 +463,35 @@ pub async fn enqueue_inter_agent_communication(
     Ok(())
 }
 
+/// Records a spawned agent's initial task under a stable spawn-call id without synchronously
+/// waking pending work.
+pub async fn enqueue_initial_agent_task(
+    sess: &Arc<Session>,
+    message_id: String,
+    communication: InterAgentCommunication,
+) -> CodexResult<()> {
+    let deferred_delivery = if communication.trigger_turn {
+        sess.input_queue
+            .defer_mailbox_delivery_for_active_turn(&sess.active_turn)
+            .await
+    } else {
+        None
+    };
+    if let Err(err) = sess
+        .input_queue
+        .enqueue_initial_task_with_id(message_id, communication)
+        .await
+    {
+        if let Some((turn_state, previous_phase)) = deferred_delivery {
+            sess.input_queue
+                .restore_mailbox_delivery_phase(turn_state.as_ref(), previous_phase)
+                .await;
+        }
+        return Err(CodexErr::InvalidRequest(err.to_string()));
+    }
+    Ok(())
+}
+
 pub async fn run_user_shell_command(sess: &Arc<Session>, sub_id: String, command: String) {
     if let Some((turn_context, cancellation_token)) =
         sess.active_turn_context_and_cancellation_token().await

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

@@ -7,6 +7,7 @@ use crate::state::TurnState;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::user_input::UserInput;
+use std::collections::HashSet;
 use std::collections::VecDeque;
 use std::error::Error;
 use std::fmt;
@@ -91,6 +92,7 @@ impl Error for MailboxQueueFull {}
 pub(crate) struct InputQueue {
     mailbox_tx: watch::Sender<()>,
     mailbox_pending_mails: Mutex<VecDeque<QueuedMailboxMessage>>,
+    initial_task_delivery_ids: Mutex<HashSet<String>>,
 }
 
 impl InputQueue {
@@ -99,6 +101,7 @@ impl InputQueue {
         Self {
             mailbox_tx,
             mailbox_pending_mails: Mutex::new(VecDeque::new()),
+            initial_task_delivery_ids: Mutex::new(HashSet::new()),
         }
     }
 
@@ -134,6 +137,36 @@ impl InputQueue {
         Ok(())
     }
 
+    /// Enqueue the spawned agent's initial task exactly once under a stable spawn-call id.
+    ///
+    /// The id remains recorded after the task drains so an acknowledgement retry cannot inject the
+    /// task a second time into a later turn. This path is used only for the one initial task of a
+    /// newly spawned child; retries reuse the same id, so the retained set normally has one entry
+    /// and cannot grow with arbitrary mailbox traffic.
+    pub(crate) async fn enqueue_initial_task_with_id(
+        &self,
+        id: String,
+        communication: InterAgentCommunication,
+    ) -> Result<(), MailboxQueueFull> {
+        let mut mails = self.mailbox_pending_mails.lock().await;
+        let mut initial_task_delivery_ids = self.initial_task_delivery_ids.lock().await;
+        if initial_task_delivery_ids.contains(&id) {
+            return Ok(());
+        }
+        if mails.len() >= MAX_MAILBOX_CONTEXT_QUEUE_ITEMS {
+            return Err(MailboxQueueFull::new(MAX_MAILBOX_CONTEXT_QUEUE_ITEMS));
+        }
+        mails.push_back(QueuedMailboxMessage {
+            id: id.clone(),
+            communication,
+        });
+        initial_task_delivery_ids.insert(id);
+        drop(initial_task_delivery_ids);
+        drop(mails);
+        self.mailbox_tx.send_replace(());
+        Ok(())
+    }
+
     pub(crate) async fn has_pending_mailbox_items(&self) -> bool {
         !self.mailbox_pending_mails.lock().await.is_empty()
     }
@@ -200,16 +233,22 @@ impl InputQueue {
 
     pub(crate) async fn drain_mailbox_input_items(&self) -> Vec<ResponseItem> {
         let mut mails = self.mailbox_pending_mails.lock().await;
+        let initial_task_delivery_ids = self.initial_task_delivery_ids.lock().await;
         let drain_count = mails.len().min(MAX_MAILBOX_CONTEXT_QUEUE_ITEMS);
         let items = mails
             .drain(..drain_count)
             .map(|mail| {
-                ResponseItem::from(
-                    MailboxContextFragment::new(mail.communication).into_response_input_item(),
-                )
+                if initial_task_delivery_ids.contains(&mail.id) {
+                    mail.communication.to_model_input_item()
+                } else {
+                    ResponseItem::from(
+                        MailboxContextFragment::new(mail.communication).into_response_input_item(),
+                    )
+                }
             })
             .collect();
         let has_more = !mails.is_empty();
+        drop(initial_task_delivery_ids);
         drop(mails);
         if has_more {
             self.mailbox_tx.send_replace(());
@@ -494,6 +533,57 @@ mod tests {
         assert!(!input_queue.has_pending_mailbox_items().await);
     }
 
+    #[tokio::test]
+    async fn input_queue_delivers_large_initial_task_exactly_once() {
+        let input_queue = InputQueue::new();
+        let initial_task = "initial task ".repeat(2_000);
+        let communication = InterAgentCommunication::new_encrypted(
+            AgentPath::root(),
+            AgentPath::try_from("/root/worker").expect("agent path"),
+            Vec::new(),
+            initial_task.clone(),
+            /*trigger_turn*/ true,
+        );
+
+        for _ in 0..2 {
+            input_queue
+                .enqueue_initial_task_with_id("spawn-call-1".to_string(), communication.clone())
+                .await
+                .expect("initial task should enqueue idempotently");
+        }
+
+        let items = input_queue.drain_mailbox_input_items().await;
+        assert_eq!(items.len(), 1);
+        let ResponseItem::AgentMessage {
+            author,
+            recipient,
+            content,
+        } = &items[0]
+        else {
+            panic!("encrypted initial task should retain the native agent message shape");
+        };
+        assert_eq!(author, "/root");
+        assert_eq!(recipient, "/root/worker");
+        let [
+            codex_protocol::models::AgentMessageInputContent::EncryptedContent {
+                encrypted_content,
+            },
+        ] = content.as_slice()
+        else {
+            panic!("initial task should retain one encrypted content item");
+        };
+        assert_eq!(encrypted_content, &initial_task);
+
+        input_queue
+            .enqueue_initial_task_with_id("spawn-call-1".to_string(), communication)
+            .await
+            .expect("delivered initial task retry should be an idempotent success");
+        assert!(
+            input_queue.drain_mailbox_input_items().await.is_empty(),
+            "delivered initial task must not be injected into a later turn"
+        );
+    }
+
     #[tokio::test]
     async fn input_queue_rejects_mailbox_when_context_queue_is_full() {
         let input_queue = InputQueue::new();

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

@@ -205,6 +205,7 @@ use codex_protocol::exec_output::StreamOutput;
 mod auth_profile_auto_switch;
 mod config_lock;
 mod handlers;
+pub(crate) use handlers::enqueue_initial_agent_task;
 pub(crate) use handlers::enqueue_inter_agent_communication;
 pub(crate) use handlers::inter_agent_communication as handle_inter_agent_communication;
 mod inject;

codex-rs/core/src/thread_manager.rs

@@ -1190,6 +1190,37 @@ impl ThreadManagerState {
         Ok(message_id)
     }
 
+    /// Delivers a spawned agent's initial task under the stable parent tool-call id.
+    ///
+    /// The input queue retains that id after delivery so a retry acknowledges success without
+    /// injecting a duplicate task into a later child turn.
+    pub(crate) async fn deliver_initial_agent_task(
+        &self,
+        thread_id: ThreadId,
+        message_id: String,
+        communication: InterAgentCommunication,
+    ) -> CodexResult<String> {
+        let thread = self.get_thread(thread_id).await?;
+        let trigger_turn = communication.trigger_turn;
+        if let Some(ops_log) = &self.ops_log
+            && let Ok(mut log) = ops_log.lock()
+        {
+            log.push((
+                thread_id,
+                Op::InterAgentCommunication {
+                    communication: communication.clone(),
+                },
+            ));
+        }
+        thread
+            .enqueue_initial_agent_task_with_id(message_id.clone(), communication)
+            .await?;
+        if trigger_turn {
+            let _ = thread.submit(Op::WakePendingWork).await;
+        }
+        Ok(message_id)
+    }
+
     /// Remove a thread from the manager by ID, returning it when present.
     pub(crate) async fn remove_thread(&self, thread_id: &ThreadId) -> Option<Arc<CodexThread>> {
         self.threads.write().await.remove(thread_id)

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

@@ -142,6 +142,7 @@ async fn handle_spawn_agent(
         SpawnAgentOptions {
             fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()),
             fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory),
+            initial_task_message_id: None,
             parent_thread_id: Some(session.thread_id),
             environments: Some(turn.environments.to_selections()),
         },