Add lifecycle hooks for inter-agent messages (#55657)

## What changed

Add `MessageLifecycleContributor` hooks to capture extension state when local agent control publishes a message and notify contributors when the recipient commits it to history, before sampling.

Carry shared delivery state through submissions, input queues, and mailbox eviction with `PendingAgentMessage`. Notify consumption at most once per delivery, release captured state when consumed or abandoned, and drop pending mail outside queue locks. Preserve the serialized message format without republishing restored history.

## Testing

Extend the mailbox eviction test to verify that publication state survives eviction and retained messages are consumed before the first resumed model sample.

GitOrigin-RevId: 9a1f6832bd48ec646a3fd179485c095e8dcc72cc

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

@@ -70,10 +70,7 @@ pub trait AgentControl: Send + Sync {
 
     /// Take queued, non-turn-starting mail in order, without loading the recipient.
     /// This in-memory operation performs no I/O; returning transfers ownership to the caller.
-    fn take_mailbox(
-        &self,
-        agent: ThreadId,
-    ) -> Vec<codex_protocol::protocol::InterAgentCommunication>;
+    fn take_mailbox(&self, agent: ThreadId) -> Vec<crate::PendingAgentMessage>;
 
     /// Observe whether unread mail is available. Subscribe before the first read so
     /// arrivals cannot be missed; notifications do not consume mail or start a turn.

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

@@ -9,6 +9,7 @@ use crate::agent::types::AgentMetadata;
 use crate::agent::types::LiveAgent;
 use crate::agent_communication::AgentCommunicationContext;
 use crate::agent_communication::AgentCommunicationKind;
+use crate::agent_communication::MessageDelivery;
 use crate::config::Config;
 use crate::config::RolloutBudgetConfig;
 use crate::context::ContextualUserFragment;
@@ -25,6 +26,7 @@ use codex_history::InitialHistory;
 use codex_history::ResumedHistory;
 use codex_history::RolloutItem;
 use codex_protocol::AgentPath;
+use codex_protocol::ResponseItemId;
 use codex_protocol::SessionId;
 use codex_protocol::ThreadId;
 use codex_protocol::error::CodexErr;
@@ -258,10 +260,20 @@ impl LocalAgentControl {
         &self,
         agent_id: ThreadId,
         state: &Arc<ThreadManagerState>,
-        communication: InterAgentCommunication,
+        mut communication: InterAgentCommunication,
         context: AgentCommunicationContext,
         start_options: TurnStartOptions,
     ) -> CodexResult<String> {
+        let observers = state.extensions.message_lifecycle_contributors();
+        let delivery = (!observers.is_empty()).then(|| {
+            let message_id = communication
+                .id
+                .take()
+                .filter(|id| !id.is_empty())
+                .unwrap_or_else(|| ResponseItemId::new("amsg"));
+            communication.id = Some(message_id.clone());
+            MessageDelivery::publish(context.sender_thread_id, message_id, observers)
+        });
         let communication_for_log =
             crate::agent_communication::logging_enabled().then(|| communication.clone());
         {
@@ -274,7 +286,7 @@ impl LocalAgentControl {
                 self.runtime.mailboxes.enqueue(
                     agent_id,
                     Some(submission_id.clone()),
-                    vec![communication],
+                    vec![crate::PendingAgentMessage::new(communication, delivery)],
                 )?;
                 if let Some(communication) = communication_for_log {
                     crate::agent_communication::emit_agent_communication_send(
@@ -309,6 +321,7 @@ impl LocalAgentControl {
                         },
                         parent_turn_id,
                         root_turn_id,
+                        delivery,
                     )
                     .await,
             )
@@ -338,6 +351,7 @@ impl LocalAgentControl {
                     Op::Interrupt,
                     /*parent_turn_id*/ None,
                     /*root_turn_id*/ None,
+                    /*message_delivery*/ None,
                 )
                 .await,
         )

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

@@ -178,10 +178,7 @@ impl AgentControl for LocalAgentControl {
         })
     }
 
-    fn take_mailbox(
-        &self,
-        agent: ThreadId,
-    ) -> Vec<codex_protocol::protocol::InterAgentCommunication> {
+    fn take_mailbox(&self, agent: ThreadId) -> Vec<crate::PendingAgentMessage> {
         self.runtime.mailboxes.take(agent)
     }
 

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

@@ -23,6 +23,7 @@ impl LocalAgentControl {
                         Op::Shutdown {},
                         /*parent_turn_id*/ None,
                         /*root_turn_id*/ None,
+                        /*message_delivery*/ None,
                     )
                     .await
             };
@@ -35,6 +36,7 @@ impl LocalAgentControl {
                     Op::Shutdown {},
                     /*parent_turn_id*/ None,
                     /*root_turn_id*/ None,
+                    /*message_delivery*/ None,
                 )
                 .await
         };

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

@@ -1,12 +1,12 @@
 //! Retains queue-only mail while local sessions are unloaded.
 //! Eviction transfers mail before releasing the recipient's residency guard.
 
+use crate::PendingAgentMessage;
 use crate::agent_communication::PENDING_MAILBOX_MESSAGES;
 use codex_diagnostics::GaugeGuard;
 use codex_protocol::ThreadId;
 use codex_protocol::error::CodexErr;
 use codex_protocol::error::Result;
-use codex_protocol::protocol::InterAgentCommunication;
 use std::collections::HashMap;
 use std::sync::Mutex;
 use std::sync::PoisonError;
@@ -28,7 +28,7 @@ struct Mailbox {
 
 struct StoredMail {
     id: Option<String>,
-    communication: InterAgentCommunication,
+    communication: PendingAgentMessage,
     _diagnostics_guard: GaugeGuard,
 }
 
@@ -37,7 +37,7 @@ impl Mailboxes {
         &self,
         agent: ThreadId,
         id: Option<String>,
-        messages: Vec<InterAgentCommunication>,
+        messages: Vec<PendingAgentMessage>,
     ) -> Result<()> {
         let mut mailboxes = self.0.lock().unwrap_or_else(PoisonError::into_inner);
         let mailbox = mailboxes
@@ -56,7 +56,7 @@ impl Mailboxes {
         Ok(())
     }
 
-    pub(super) fn take(&self, agent: ThreadId) -> Vec<InterAgentCommunication> {
+    pub(super) fn take(&self, agent: ThreadId) -> Vec<PendingAgentMessage> {
         let mut mailboxes = self.0.lock().unwrap_or_else(PoisonError::into_inner);
         let Some(mailbox) = mailboxes
             .as_mut()
@@ -94,18 +94,20 @@ impl Mailboxes {
     }
 
     pub(super) fn remove(&self, agent: ThreadId) {
-        if let Some(mailboxes) = self
+        // Dropping pending mail can run extension destructors. Release the lock first.
+        let mailbox = self
             .0
             .lock()
             .unwrap_or_else(PoisonError::into_inner)
             .as_mut()
-        {
-            mailboxes.remove(&agent);
-        }
+            .and_then(|mailboxes| mailboxes.remove(&agent));
+        drop(mailbox);
     }
 
     pub(super) fn close(&self) {
-        if let Some(mailboxes) = self.0.lock().unwrap_or_else(PoisonError::into_inner).take() {
+        // Detach under the lock, then notify readers and drop pending mail outside it.
+        let mailboxes = self.0.lock().unwrap_or_else(PoisonError::into_inner).take();
+        if let Some(mailboxes) = mailboxes {
             for mailbox in mailboxes.values() {
                 mailbox.state.send_replace(false);
             }

codex-rs/core/src/agent_communication.rs

@@ -1,5 +1,19 @@
+//! Direct-message diagnostics and delivery ownership across queues and eviction.
+//! Only history commitment consumes mail; dropping its last owner releases its attachments.
+
+use codex_extension_api::ExtensionData;
+use codex_extension_api::MessageLifecycleContributor;
+use codex_extension_api::MessageLifecycleInput;
+use codex_protocol::ResponseItemId;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::InterAgentCommunication;
+use serde::Deserialize;
+use serde::Serialize;
+use std::fmt;
+use std::ops::Deref;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::PoisonError;
 
 pub(crate) static PENDING_MAILBOX_MESSAGES: codex_diagnostics::Gauge =
     codex_diagnostics::Gauge::new("core.mailbox.pending");
@@ -28,7 +42,7 @@ impl AgentCommunicationKind {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub(crate) struct AgentCommunicationContext {
     kind: AgentCommunicationKind,
-    sender_thread_id: ThreadId,
+    pub(crate) sender_thread_id: ThreadId,
 }
 
 impl AgentCommunicationContext {
@@ -79,3 +93,92 @@ pub(crate) fn emit_agent_communication_receive(communication_id: &str) {
         "agent communication"
     );
 }
+
+/// Pending mail with shared ownership of its captured publication state.
+/// Serialization preserves the protocol shape; restored historical mail is not republished.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct PendingAgentMessage {
+    communication: InterAgentCommunication,
+    #[serde(skip)]
+    pub(crate) delivery: Option<MessageDelivery>,
+}
+
+impl From<InterAgentCommunication> for PendingAgentMessage {
+    fn from(communication: InterAgentCommunication) -> Self {
+        Self::new(communication, /*delivery*/ None)
+    }
+}
+
+impl Deref for PendingAgentMessage {
+    type Target = InterAgentCommunication;
+    fn deref(&self) -> &Self::Target {
+        &self.communication
+    }
+}
+
+impl PartialEq for PendingAgentMessage {
+    fn eq(&self, other: &Self) -> bool {
+        self.communication == other.communication
+    }
+}
+
+impl PendingAgentMessage {
+    pub(crate) fn new(
+        communication: InterAgentCommunication,
+        delivery: Option<MessageDelivery>,
+    ) -> Self {
+        Self {
+            communication,
+            delivery,
+        }
+    }
+}
+
+/// The publication state retained until this delivery is consumed or abandoned.
+struct MessagePublication {
+    input: MessageLifecycleInput,
+    observers: Box<[Arc<dyn MessageLifecycleContributor>]>,
+}
+
+/// Clones follow one logical delivery through queues and share its consumption state.
+#[derive(Clone)]
+pub(crate) struct MessageDelivery(Arc<Mutex<Option<MessagePublication>>>);
+
+impl MessageDelivery {
+    pub(crate) fn publish(
+        sender: ThreadId,
+        message_id: ResponseItemId,
+        observers: &[Arc<dyn MessageLifecycleContributor>],
+    ) -> Self {
+        let input = MessageLifecycleInput {
+            sender,
+            data: ExtensionData::new(message_id.to_string()),
+            message_id,
+        };
+        for observer in observers {
+            observer.on_publish(&input);
+        }
+        let publication = MessagePublication {
+            input,
+            observers: observers.into(),
+        };
+        Self(Arc::new(Mutex::new(Some(publication))))
+    }
+
+    /// Commit this delivery and release its publication state, even if clones remain.
+    pub(crate) fn consumed(&self, recipient: ThreadId) {
+        let publication = self.0.lock().unwrap_or_else(PoisonError::into_inner).take();
+        if let Some(publication) = publication {
+            for observer in &publication.observers {
+                observer.on_consumed(&publication.input, recipient);
+            }
+        }
+    }
+}
+
+impl fmt::Debug for MessageDelivery {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("MessageDelivery").finish_non_exhaustive()
+    }
+}

codex-rs/core/src/codex_delegate.rs

@@ -311,6 +311,7 @@ pub(crate) async fn run_codex_thread_one_shot(
                         parent_turn_id: None,
                         root_turn_id: None,
                         residency_guard: None,
+                        message_delivery: None,
                     })
                     .await;
                 child_cancel.cancel();

codex-rs/core/src/codex_delegate_tests.rs

@@ -163,6 +163,7 @@ async fn forward_ops_preserves_submission_trace_context() {
         parent_turn_id: Some("parent-turn".to_string()),
         root_turn_id: Some("root-turn".to_string()),
         residency_guard: None,
+        message_delivery: None,
     };
     tx_ops.send(submission).await.unwrap();
     drop(tx_ops);

codex-rs/core/src/codex_thread.rs

@@ -404,7 +404,7 @@ impl CodexThread {
         self.io
             .submit_with_trace(
                 op, trace, /*parent_turn_id*/ None, /*root_turn_id*/ None,
-                /*residency_guard*/ None,
+                /*residency_guard*/ None, /*message_delivery*/ None,
             )
             .await
     }
@@ -550,6 +550,7 @@ impl CodexThread {
                 parent_turn_id: None,
                 root_turn_id: None,
                 residency_guard: None,
+                message_delivery: None,
             })
             .await
             .map_err(|_| CodexErr::Fatal("thread session has stopped".to_string()))?;

codex-rs/core/src/lib.rs

@@ -77,6 +77,7 @@ pub use agent::types::SpawnAgentForkMode;
 pub use agent::types::SpawnAgentOptions;
 pub use rollout_budget::RolloutBudgetReminder;
 mod agent_communication;
+pub use agent_communication::PendingAgentMessage;
 mod agent_message_board;
 pub use agent_message_board::install_agent_message_board;
 mod attestation;

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

@@ -34,7 +34,6 @@ use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::GuardianAssessmentEvent;
 use codex_protocol::protocol::GuardianAssessmentStatus;
-use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::protocol::Op;
 use codex_protocol::protocol::RealtimeConversationListVoicesResponseEvent;
 use codex_protocol::protocol::RealtimeVoicesList;
@@ -83,7 +82,7 @@ pub async fn realtime_conversation_list_voices(sess: &Session, sub_id: String) {
 pub async fn inter_agent_communication(
     sess: &Arc<Session>,
     sub_id: String,
-    communication: InterAgentCommunication,
+    communication: crate::PendingAgentMessage,
     start_options: codex_protocol::turn_input::TurnStartOptions,
 ) {
     let trigger_turn = communication.trigger_turn;
@@ -657,6 +656,8 @@ pub(super) async fn submission_loop(
                     communication,
                     start_options,
                 } => {
+                    let communication =
+                        crate::PendingAgentMessage::new(communication, sub.message_delivery);
                     inter_agent_communication(&sess, sub.id.clone(), communication, start_options)
                         .await;
                     false

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

@@ -44,7 +44,7 @@ pub enum TurnInput {
     // Preserve the existing serialized format while carrying injection API metadata
     // through the in-memory queue.
     ResponseItem(#[serde(with = "turn_input_response_item")] ResponseItemEnvelope),
-    InterAgentCommunication(InterAgentCommunication),
+    InterAgentCommunication(crate::PendingAgentMessage),
 }
 
 mod turn_input_response_item {
@@ -101,7 +101,7 @@ pub(crate) struct InputQueue {
 }
 
 pub(crate) struct PendingMailboxCommunication {
-    pub(crate) communication: InterAgentCommunication,
+    pub(crate) communication: crate::PendingAgentMessage,
     start_options: TurnStartOptions,
     _diagnostics_guard: GaugeGuard,
 }
@@ -199,14 +199,14 @@ impl InputQueue {
         turn_state
             .pending_input
             .items
-            .push(TurnInput::InterAgentCommunication(communication));
+            .push(TurnInput::InterAgentCommunication(communication.into()));
         self.activity_tx.send_replace(InputQueueActivity::Mailbox);
         true
     }
 
     pub(crate) async fn enqueue_mailbox_communication(
         &self,
-        communication: InterAgentCommunication,
+        communication: crate::PendingAgentMessage,
         start_options: TurnStartOptions,
     ) {
         let mut pending = self.mailbox_pending_mails.lock().await;
@@ -318,7 +318,10 @@ impl InputQueue {
     pub(crate) async fn clear_pending(&self, active_turn: &ActiveTurn) {
         let mut turn_state = active_turn.turn_state.lock().await;
         turn_state.clear_pending_waiters();
-        turn_state.pending_input = TurnInputQueue::default();
+        // Pending mail can own extension attachments whose destructors need the lock released.
+        let pending_input = std::mem::take(&mut turn_state.pending_input);
+        drop(turn_state);
+        drop(pending_input);
     }
 
     pub(crate) async fn defer_mailbox_delivery_to_next_turn(
@@ -586,7 +589,7 @@ mod tests {
             /*trigger_turn*/ false,
         );
         input_queue
-            .enqueue_mailbox_communication(mail_one, Default::default())
+            .enqueue_mailbox_communication(mail_one.into(), Default::default())
             .await;
         let mail_two = make_mail(
             AgentPath::root(),
@@ -595,7 +598,7 @@ mod tests {
             /*trigger_turn*/ false,
         );
         input_queue
-            .enqueue_mailbox_communication(mail_two, Default::default())
+            .enqueue_mailbox_communication(mail_two.into(), Default::default())
             .await;
 
         activity_rx.changed().await.expect("mailbox update");
@@ -676,7 +679,7 @@ mod tests {
         input_queue
             .extend_pending_input_for_turn_state(
                 &turn_state,
-                vec![TurnInput::InterAgentCommunication(communication)],
+                vec![TurnInput::InterAgentCommunication(communication.into())],
             )
             .await;
         assert_eq!(
@@ -720,17 +723,17 @@ mod tests {
         );
 
         input_queue
-            .enqueue_mailbox_communication(mail_one.clone(), Default::default())
+            .enqueue_mailbox_communication(mail_one.clone().into(), Default::default())
             .await;
         input_queue
-            .enqueue_mailbox_communication(mail_two.clone(), Default::default())
+            .enqueue_mailbox_communication(mail_two.clone().into(), Default::default())
             .await;
 
         assert_eq!(
             input_queue.drain_mailbox_input_items().await.0,
             vec![
-                TurnInput::InterAgentCommunication(mail_one),
-                TurnInput::InterAgentCommunication(mail_two)
+                TurnInput::InterAgentCommunication(mail_one.into()),
+                TurnInput::InterAgentCommunication(mail_two.into())
             ]
         );
         assert!(!input_queue.has_pending_mailbox_items().await);
@@ -765,7 +768,8 @@ mod tests {
             for (trigger_turn, parent_turn_id, root_turn_id) in pending_mails {
                 input_queue
                     .enqueue_mailbox_communication(
-                        make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
+                        make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn)
+                            .into(),
                         TurnStartOptions {
                             parent_turn_id: parent_turn_id.map(str::to_string),
                             root_turn_id: root_turn_id.map(str::to_string),
@@ -796,7 +800,8 @@ mod tests {
             ] {
                 input_queue
                     .enqueue_mailbox_communication(
-                        make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
+                        make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn)
+                            .into(),
                         TurnStartOptions {
                             cyber_access_program: program,
                             ..Default::default()
@@ -820,7 +825,7 @@ mod tests {
             /*trigger_turn*/ false,
         );
         input_queue
-            .enqueue_mailbox_communication(queued_mail, Default::default())
+            .enqueue_mailbox_communication(queued_mail.into(), Default::default())
             .await;
         assert!(!input_queue.has_trigger_turn_mailbox_items().await);
 
@@ -831,7 +836,7 @@ mod tests {
             /*trigger_turn*/ true,
         );
         input_queue
-            .enqueue_mailbox_communication(trigger_mail, Default::default())
+            .enqueue_mailbox_communication(trigger_mail.into(), Default::default())
             .await;
         assert!(input_queue.has_trigger_turn_mailbox_items().await);
     }

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

@@ -145,7 +145,6 @@ use codex_protocol::permissions::NetworkSandboxPolicy;
 use codex_protocol::protocol::FileChange;
 use codex_protocol::protocol::HasLegacyEvent;
 use codex_protocol::protocol::HistoryPosition;
-use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::protocol::ItemCompletedEvent;
 use codex_protocol::protocol::ItemStartedEvent;
 use codex_protocol::protocol::MULTI_AGENT_MODE_OPEN_TAG;
@@ -1060,7 +1059,7 @@ impl SessionIo {
     ) -> CodexResult<String> {
         self.submit_with_trace(
             op, /*trace*/ None, /*parent_turn_id*/ None, /*root_turn_id*/ None,
-            /*residency_guard*/ None,
+            /*residency_guard*/ None, /*message_delivery*/ None,
         )
         .await
     }
@@ -1072,6 +1071,7 @@ impl SessionIo {
         parent_turn_id: Option<String>,
         root_turn_id: Option<String>,
         residency_guard: Option<tokio::sync::OwnedRwLockReadGuard<()>>,
+        message_delivery: Option<crate::agent_communication::MessageDelivery>,
     ) -> CodexResult<String> {
         let id = new_submission_id();
         let WithTurnExtensionData {
@@ -1086,6 +1086,7 @@ impl SessionIo {
             parent_turn_id,
             root_turn_id,
             residency_guard,
+            message_delivery,
         };
         self.submit_with_id(sub).await?;
         Ok(id)
@@ -1131,6 +1132,7 @@ impl SessionIo {
             parent_turn_id: None,
             root_turn_id: None,
             residency_guard: None,
+            message_delivery: None,
         })
         .await?;
         reply_rx.await.unwrap_or(Err(CodexErr::InternalAgentDied))
@@ -1157,6 +1159,7 @@ impl SessionIo {
             parent_turn_id: None,
             root_turn_id: None,
             residency_guard: None,
+            message_delivery: None,
         })
         .await?;
         reply_rx.await.unwrap_or(Err(CodexErr::InternalAgentDied))
@@ -4093,8 +4096,9 @@ impl Session {
         &self,
         turn_context: &TurnContext,
         model_info: &ModelInfo,
-        communication: InterAgentCommunication,
+        communication: impl Into<crate::PendingAgentMessage>,
     ) {
+        let communication = communication.into();
         let mut response_item = communication.to_model_input_item();
         if let ResponseItem::AgentMessage {
             content,
@@ -4161,6 +4165,9 @@ impl Session {
                 model_info.truncation_policy.into(),
             );
         }
+        if let Some(delivery) = &communication.delivery {
+            delivery.consumed(self.thread_id);
+        }
         self.persist_rollout_items(&[
             RolloutItem::InterAgentCommunicationMetadata {
                 trigger_turn: communication.trigger_turn,

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

@@ -513,6 +513,7 @@ fn settings_submission(
             parent_turn_id: None,
             root_turn_id: None,
             residency_guard: None,
+            message_delivery: None,
         },
         receiver,
     )

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

@@ -17,4 +17,6 @@ pub(crate) struct Submission {
     pub root_turn_id: Option<String>,
     /// Keeps a V2 recipient resident until this submission is handled or dropped.
     pub residency_guard: Option<OwnedRwLockReadGuard<()>>,
+    /// Owns delivery until dispatch; failed or abandoned submissions discard it.
+    pub message_delivery: Option<crate::agent_communication::MessageDelivery>,
 }

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

@@ -6176,6 +6176,7 @@ async fn standalone_settings_invalidate_continuation_before_delivering_acceptanc
             parent_turn_id: None,
             root_turn_id: None,
             residency_guard: None,
+            message_delivery: None,
         })
         .await
         .expect("submit settings");
@@ -8134,6 +8135,7 @@ async fn submit_with_trace_captures_current_span_trace_context() {
             /*parent_turn_id*/ None,
             /*root_turn_id*/ None,
             /*residency_guard*/ None,
+            /*message_delivery*/ None,
         )
         .await
         .expect("submit should succeed");
@@ -8208,6 +8210,7 @@ fn submission_dispatch_span_prefers_submission_trace_context() {
             parent_turn_id: None,
             root_turn_id: None,
             residency_guard: None,
+            message_delivery: None,
             trace: Some(submission_trace),
         })
     });
@@ -8238,6 +8241,7 @@ fn submission_dispatch_span_uses_debug_for_realtime_audio() {
         parent_turn_id: None,
         root_turn_id: None,
         residency_guard: None,
+        message_delivery: None,
         trace: None,
     });
 
@@ -8598,6 +8602,7 @@ async fn spawn_task_turn_span_inherits_dispatch_trace_context() {
         parent_turn_id: None,
         root_turn_id: None,
         residency_guard: None,
+        message_delivery: None,
         trace: Some(submission_trace.clone()),
     });
     let dispatch_span_id = dispatch_span.context().span().span_context().span_id();
@@ -12716,7 +12721,8 @@ async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() {
                 Vec::new(),
                 "pending trigger".to_string(),
                 /*trigger_turn*/ true,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;
@@ -12798,7 +12804,7 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
         .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
         .await;
     sess.input_queue
-        .enqueue_mailbox_communication(communication.clone(), Default::default())
+        .enqueue_mailbox_communication(communication.clone().into(), Default::default())
         .await;
 
     assert!(
@@ -12817,7 +12823,7 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
 
     assert_eq!(
         (sess.input_queue.get_pending_input(&sess.active_turn).await).0,
-        vec![TurnInput::InterAgentCommunication(communication)],
+        vec![TurnInput::InterAgentCommunication(communication.into())],
     );
 }
 
@@ -12845,7 +12851,8 @@ async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
                 Vec::new(),
                 "late trigger update".to_string(),
                 /*trigger_turn*/ true,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;
@@ -12891,7 +12898,7 @@ async fn active_turn_keeps_first_root_when_mail_coalesces(inherited_root: Option
     {
         sess.input_queue
             .enqueue_mailbox_communication(
-                communication,
+                communication.into(),
                 codex_protocol::turn_input::TurnStartOptions {
                     parent_turn_id: Some(parent_turn_id.to_string()),
                     root_turn_id: Some(root_turn_id.to_string()),
@@ -12917,8 +12924,8 @@ async fn active_turn_keeps_first_root_when_mail_coalesces(inherited_root: Option
     assert_eq!(
         (sess.input_queue.get_pending_input(&sess.active_turn).await).0,
         vec![
-            TurnInput::InterAgentCommunication(first),
-            TurnInput::InterAgentCommunication(second),
+            TurnInput::InterAgentCommunication(first.into()),
+            TurnInput::InterAgentCommunication(second.into()),
         ]
     );
     assert_eq!(
@@ -12954,7 +12961,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
         .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
         .await;
     sess.input_queue
-        .enqueue_mailbox_communication(communication.clone(), Default::default())
+        .enqueue_mailbox_communication(communication.clone().into(), Default::default())
         .await;
     let submission = submit_steer_only(
         &sess,
@@ -12981,7 +12988,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
                 }],
                 client_id: None,
             },
-            TurnInput::InterAgentCommunication(communication),
+            TurnInput::InterAgentCommunication(communication.into()),
         ],
     );
 }
@@ -13010,7 +13017,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
         .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
         .await;
     sess.input_queue
-        .enqueue_mailbox_communication(communication.clone(), Default::default())
+        .enqueue_mailbox_communication(communication.clone().into(), Default::default())
         .await;
     let submission = submit_steer_only(
         &sess,
@@ -13041,7 +13048,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
                 }],
                 client_id: None,
             },
-            TurnInput::InterAgentCommunication(communication),
+            TurnInput::InterAgentCommunication(communication.into()),
         ],
     );
 }
@@ -13070,7 +13077,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
         .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
         .await;
     sess.input_queue
-        .enqueue_mailbox_communication(communication.clone(), Default::default())
+        .enqueue_mailbox_communication(communication.clone().into(), Default::default())
         .await;
 
     let item = ResponseItem::FunctionCall {
@@ -13101,7 +13108,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
     assert!(output.tool_future.is_some());
     assert_eq!(
         (sess.input_queue.get_pending_input(&sess.active_turn).await).0,
-        vec![TurnInput::InterAgentCommunication(communication)],
+        vec![TurnInput::InterAgentCommunication(communication.into())],
     );
 }
 

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

@@ -993,7 +993,7 @@ async fn pending_turn_input(
             TurnInput::ResponseItem(item.into())
         }
         SubmittedTurnInput::InterAgentCommunication(communication) => {
-            TurnInput::InterAgentCommunication(communication)
+            TurnInput::InterAgentCommunication(communication.into())
         }
     };
     if retain

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

@@ -831,7 +831,8 @@ async fn start_only_rejects_pending_trigger_turn_without_injecting() {
                 Vec::new(),
                 "pending trigger".to_string(),
                 /*trigger_turn*/ true,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;
@@ -887,14 +888,14 @@ async fn agent_message_wake_reservation_preserves_existing_boundaries(
     );
     session
         .input_queue
-        .enqueue_mailbox_communication(communication.clone(), Default::default())
+        .enqueue_mailbox_communication(communication.clone().into(), Default::default())
         .await;
 
     assert!(session.reserve_pending_work_turn().await.is_none());
     assert!(session.active_turn.lock().await.is_none());
     assert_eq!(
         session.input_queue.drain_mailbox_input_items().await.0,
-        vec![TurnInput::InterAgentCommunication(communication)]
+        vec![TurnInput::InterAgentCommunication(communication.into())]
     );
 }
 

codex-rs/core/src/thread_manager.rs

@@ -452,7 +452,7 @@ pub(crate) struct ThreadManagerState {
     plugins_manager: Arc<PluginsManager>,
     mcp_manager: Arc<McpManager>,
     code_mode_session_provider: Arc<dyn CodeModeSessionProvider>,
-    extensions: Arc<ExtensionRegistry<Config>>,
+    pub(crate) extensions: Arc<ExtensionRegistry<Config>>,
     user_instructions_provider: Arc<dyn UserInstructionsProvider>,
     image_store: Arc<dyn AttachmentStore>,
     thread_store: Arc<dyn ThreadStore>,
@@ -1816,6 +1816,7 @@ impl ThreadManagerState {
         op: Op,
         parent_turn_id: Option<String>,
         root_turn_id: Option<String>,
+        message_delivery: Option<crate::agent_communication::MessageDelivery>,
     ) -> CodexResult<String> {
         let thread = self.get_thread(thread_id).await?;
         let residency_guard = if matches!(op, Op::InterAgentCommunication { .. }) {
@@ -1842,6 +1843,7 @@ impl ThreadManagerState {
                 parent_turn_id,
                 root_turn_id,
                 residency_guard,
+                message_delivery,
             )
             .await
     }

codex-rs/core/src/tools/handlers/multi_agents_tests.rs

@@ -3037,7 +3037,8 @@ async fn multi_agent_v2_wait_agent_accepts_timeout_only_argument() {
                 Vec::new(),
                 "hello from worker".to_string(),
                 /*trigger_turn*/ false,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;
@@ -3537,7 +3538,8 @@ async fn multi_agent_v2_wait_agent_returns_summary_for_mailbox_activity() {
                 Vec::new(),
                 "completed".to_string(),
                 /*trigger_turn*/ false,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;
@@ -3613,7 +3615,8 @@ async fn multi_agent_v2_wait_agent_returns_for_already_queued_mail() {
                 Vec::new(),
                 "already queued".to_string(),
                 /*trigger_turn*/ false,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;
@@ -3715,7 +3718,8 @@ async fn multi_agent_v2_wait_agent_wakes_on_any_mailbox_notification() {
                 Vec::new(),
                 "from worker b".to_string(),
                 /*trigger_turn*/ false,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;
@@ -3806,7 +3810,8 @@ async fn multi_agent_v2_wait_agent_does_not_return_completed_content() {
                 Vec::new(),
                 "sensitive child output".to_string(),
                 /*trigger_turn*/ false,
-            ),
+            )
+            .into(),
             Default::default(),
         )
         .await;

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

@@ -15,6 +15,7 @@ use codex_core::DeliveryReceipt;
 use codex_core::ForkSnapshot;
 use codex_core::GuardianRootSnapshot;
 use codex_core::LiveAgent;
+use codex_core::PendingAgentMessage;
 use codex_core::RolloutBudgetReminder;
 use codex_core::SendRequest;
 use codex_core::SpawnRequest;
@@ -53,7 +54,7 @@ use wiremock::MockServer;
 struct TestAgentControl {
     thread_id: ThreadId,
     service_tier: Mutex<Option<String>>,
-    mail: Mutex<Vec<codex_protocol::protocol::InterAgentCommunication>>,
+    mail: Mutex<Vec<PendingAgentMessage>>,
     mailbox_state: tokio::sync::watch::Sender<bool>,
     agents: Mutex<HashMap<String, ThreadId>>,
 }
@@ -104,10 +105,7 @@ impl AgentControl for TestAgentControl {
         Box::pin(async { Err(CodexErr::InvalidRequest("host send rejection".to_string())) })
     }
 
-    fn take_mailbox(
-        &self,
-        _agent: ThreadId,
-    ) -> Vec<codex_protocol::protocol::InterAgentCommunication> {
+    fn take_mailbox(&self, _agent: ThreadId) -> Vec<PendingAgentMessage> {
         let messages = std::mem::take(&mut *self.mail.lock().expect("mail lock"));
         self.mailbox_state.send_if_modified(std::mem::take);
         messages
@@ -287,13 +285,14 @@ async fn host_threads_preserve_lineage_settings_and_resume_routing() -> anyhow::
         (controller.identity(), Some(root_id))
     );
     assert_eq!(controller.service_tier(), Some("priority".to_string()));
-    let unread_mail = vec![codex_protocol::protocol::InterAgentCommunication::new(
+    let message = codex_protocol::protocol::InterAgentCommunication::new(
         AgentPath::root(),
         AgentPath::root().join("worker").expect("valid child path"),
         Vec::new(),
         "Retain this in the host mailbox".into(),
         /*trigger_turn*/ false,
-    )];
+    );
+    let unread_mail = vec![message.into()];
     *controller.mail.lock().expect("mail lock") = unread_mail.clone();
     let error = test
         .thread_manager
@@ -591,7 +590,8 @@ async fn host_mailbox_notifies_waiting_agent_and_delivers_once() -> anyhow::Resu
             Vec::new(),
             "Note from the host mailbox".into(),
             /*trigger_turn*/ false,
-        ),
+        )
+        .into(),
     );
     controller.mailbox_state.send_replace(true);
     wait_for_event(&test.codex, |event| {
@@ -658,7 +658,11 @@ async fn agent_mail_wakes_completed_root_when_enabled(
         /*trigger_turn*/ false,
     );
     if via_host_mailbox {
-        controller.mail.lock().expect("mail lock").push(message);
+        controller
+            .mail
+            .lock()
+            .expect("mail lock")
+            .push(message.into());
         controller.mailbox_state.send_replace(true);
     } else {
         test.codex

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

@@ -5,15 +5,33 @@ use crate::suite::settings_commits::COMMITTED_MODEL;
 use crate::suite::settings_commits::PauseAfterCommit;
 use codex_core::config::Config;
 use codex_extension_api::ExtensionRegistryBuilder;
+use codex_extension_api::MessageLifecycleContributor;
+use codex_extension_api::MessageLifecycleInput;
+use codex_protocol::ResponseItemId;
+use codex_protocol::ThreadId;
 use codex_protocol::protocol::AgentStatus;
 use codex_protocol::protocol::Op;
 use core_test_support::ThreadIdle;
 use pretty_assertions::assert_eq;
+use std::collections::HashSet;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::sync::mpsc;
 use tokio::sync::oneshot;
 
+#[derive(Default)]
+struct Messages(Mutex<HashSet<ResponseItemId>>);
+impl MessageLifecycleContributor for Messages {
+    fn on_publish(&self, input: &MessageLifecycleInput) {
+        let mut pending = self.0.lock().expect("messages lock");
+        assert!(pending.insert(input.message_id.clone()));
+    }
+    fn on_consumed(&self, input: &MessageLifecycleInput, _recipient: ThreadId) {
+        let mut pending = self.0.lock().expect("messages lock");
+        assert!(pending.remove(&input.message_id));
+    }
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn idle_mail_survives_eviction_and_preserves_followup_order() -> Result<()> {
     const MAIL: &str = "Keep this note for later";
@@ -37,6 +55,8 @@ async fn idle_mail_survives_eviction_and_preserves_followup_order() -> Result<()
         gate: Mutex::new(Some((entered_tx, release_rx))),
     }));
     extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle));
+    let messages = Arc::new(Messages::default());
+    extensions.message_lifecycle_contributor(messages.clone());
     let test = test_codex()
         .with_model("gpt-5.6-sol")
         .with_extensions(Arc::new(extensions.build()))
@@ -106,6 +126,8 @@ async fn idle_mail_survives_eviction_and_preserves_followup_order() -> Result<()
     ThreadIdle::wait(&test.codex).await;
     assert!(test.thread_manager.get_thread(first_id).await.is_err());
 
+    let retained = messages.0.lock().unwrap().clone();
+    assert_eq!(retained.len(), 2);
     // Reload without starting a turn, then hold dispatch while both tools submit.
     // The older queued notes must precede the follow-up, and the newer note must follow it.
     test.thread_manager
@@ -130,10 +152,18 @@ async fn idle_mail_survives_eviction_and_preserves_followup_order() -> Result<()
         json!({ "target": "first", "message": FOLLOWUP }),
     )
     .await;
+    let (at_sampling, pending_at_sampling) = mpsc::channel();
     mount_sse_once_match(
         &server,
-        |request: &wiremock::Request| {
-            body_contains(request, FOLLOWUP) && !has_function_call_output(request, "followup")
+        move |request: &wiremock::Request| {
+            let matches =
+                body_contains(request, FOLLOWUP) && !has_function_call_output(request, "followup");
+            if matches {
+                at_sampling
+                    .send(messages.0.lock().unwrap().clone())
+                    .unwrap();
+            }
+            matches
         },
         sse(vec![
             ev_response_created("worker-wait"),
@@ -171,6 +201,10 @@ async fn idle_mail_survives_eviction_and_preserves_followup_order() -> Result<()
     release_tx.send(())?;
     wait_for_event(&resumed, |event| matches!(event, EventMsg::TurnComplete(_))).await;
     ThreadIdle::wait(&resumed).await;
+    assert!(
+        pending_at_sampling.try_recv()?.is_disjoint(&retained),
+        "both retained publications must be consumed before the first resumed sample"
+    );
     let request = delivered
         .requests()
         .into_iter()

codex-rs/ext/extension-api/src/lib.rs

@@ -1,5 +1,6 @@
 mod capabilities;
 mod contributors;
+mod message_lifecycle;
 mod registry;
 mod session_isolation;
 mod state;
@@ -7,6 +8,8 @@ mod tool_policy;
 mod turn_admission;
 mod user_instructions;
 
+pub use message_lifecycle::MessageLifecycleContributor;
+pub use message_lifecycle::MessageLifecycleInput;
 pub use session_isolation::IsolatedSessionExtensions;
 pub use session_isolation::SessionIsolation;
 pub use tool_policy::ToolPolicy;

codex-rs/ext/extension-api/src/message_lifecycle.rs

@@ -0,0 +1,23 @@
+//! Capture publication state once; consume it independently for each delivery.
+
+use crate::ExtensionData;
+use codex_protocol::ResponseItemId;
+use codex_protocol::ThreadId;
+
+/// State shared by all deliveries of a publication, including recipients resolved later.
+pub struct MessageLifecycleInput {
+    pub message_id: ResponseItemId,
+    pub sender: ThreadId,
+    /// Capture contributor-specific state at publication; read it at consumption.
+    pub data: ExtensionData,
+}
+
+/// Observes messages published through local agent control, where the sender is known.
+/// Callbacks are synchronous and must not block. Replaying history or submitting raw
+/// protocol input does not create a publication.
+pub trait MessageLifecycleContributor: Send + Sync {
+    /// Called once before the message is exposed to any recipient.
+    fn on_publish(&self, _input: &MessageLifecycleInput) {}
+    /// Called at most once per logical delivery, at history commitment before sampling.
+    fn on_consumed(&self, _input: &MessageLifecycleInput, _recipient: ThreadId) {}
+}

codex-rs/ext/extension-api/src/registry.rs

@@ -5,6 +5,7 @@ use crate::ConfigContributor;
 use crate::ContextContributor;
 use crate::ExtensionEventSink;
 use crate::McpServerContributor;
+use crate::MessageLifecycleContributor;
 use crate::ModelRequestContributor;
 use crate::NoopExtensionEventSink;
 use crate::SkillInvocationContributor;
@@ -40,6 +41,7 @@ impl<C: Sync> Default for ExtensionRegistryBuilder<C> {
                 tool_contributors: Vec::new(),
                 tool_lifecycle_contributors: Vec::new(),
                 model_request_contributors: Vec::new(),
+                message_observers: Vec::new(),
                 turn_item_contributors: Vec::new(),
             },
         }
@@ -144,6 +146,11 @@ impl<C: Sync> ExtensionRegistryBuilder<C> {
         self.registry.turn_item_contributors.push(contributor);
     }
 
+    /// Registers a synchronous direct-message observer.
+    pub fn message_lifecycle_contributor(&mut self, hook: Arc<dyn MessageLifecycleContributor>) {
+        self.registry.message_observers.push(hook);
+    }
+
     /// Finishes construction and returns the immutable registry.
     pub fn build(self) -> ExtensionRegistry<C> {
         self.registry
@@ -165,6 +172,7 @@ pub struct ExtensionRegistry<C: Sync> {
     tool_contributors: Vec<Arc<dyn ToolContributor>>,
     tool_lifecycle_contributors: Vec<Arc<dyn ToolLifecycleContributor>>,
     model_request_contributors: Vec<Arc<dyn ModelRequestContributor>>,
+    message_observers: Vec<Arc<dyn MessageLifecycleContributor>>,
     turn_item_contributors: Vec<Arc<dyn TurnItemContributor>>,
     approval_review_contributors: Vec<Arc<dyn ApprovalReviewContributor>>,
 }
@@ -187,6 +195,7 @@ impl<C: Sync> ExtensionRegistry<C> {
                 tool_contributors: self.tool_contributors.clone(),
                 tool_lifecycle_contributors: self.tool_lifecycle_contributors.clone(),
                 model_request_contributors: self.model_request_contributors.clone(),
+                message_observers: self.message_observers.clone(),
                 turn_item_contributors: self.turn_item_contributors.clone(),
                 approval_review_contributors: self.approval_review_contributors.clone(),
             },
@@ -276,6 +285,11 @@ impl<C: Sync> ExtensionRegistry<C> {
         &self.tool_contributors
     }
 
+    /// Returns direct-message observers in registration order.
+    pub fn message_lifecycle_contributors(&self) -> &[Arc<dyn MessageLifecycleContributor>] {
+        &self.message_observers
+    }
+
     /// Returns response interceptor contributors in registration order.
     pub fn model_request_contributors(&self) -> &[Arc<dyn ModelRequestContributor>] {
         &self.model_request_contributors