Route message-board notifications through agent mailboxes (#59139)

## Why

Board notifications were skipped when recipients had no active turn accepting mailbox delivery. Queue them through the ordinary agent-message path so idle recipients can receive them in a later turn.

## What changed

- Submit board notices as `Op::InterAgentCommunication` with `trigger_turn` disabled, preserving normal mailbox scheduling without reopening finalized answers.
- Hold a residency guard through dispatch of submitted agent messages to coordinate delivery with agent eviction.
- Update board tool descriptions to describe the host's notification queueing and wake policy.

## Testing

Update local and HTTP board scenarios to cover queued idle notices, delivery after a finalized answer, and mailbox delivery before a pending tool call. Extend residency coverage to verify that submission waits for the eviction gate and queued messages survive eviction.

GitOrigin-RevId: a77137fc40932f5903852e14742483844e07c9bc

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

@@ -171,15 +171,15 @@ impl LocalAgentRuntime {
     /// Pins and touches the registered runtime without waiting for unrelated eviction.
     pub(crate) async fn pin_v2_residency(
         &self,
-        state: &ThreadManagerState,
-        thread: &Arc<CodexThread>,
+        thread: &CodexThread,
     ) -> CodexResult<Option<tokio::sync::OwnedRwLockReadGuard<()>>> {
         if !is_resident_candidate(thread) {
             return Ok(None);
         }
         let guard = Arc::clone(&thread.residency_gate).read_owned().await;
         let thread_id = thread.session.thread_id;
-        if !Arc::ptr_eq(thread, &state.get_thread(thread_id).await?) {
+        let registered = self.upgrade()?.get_thread(thread_id).await?;
+        if !std::ptr::eq(thread, registered.as_ref()) {
             return Err(CodexErr::ThreadNotFound(thread_id));
         }
         self.residency.touch(thread_id);

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

@@ -7,11 +7,15 @@ use crate::config::test_config;
 use crate::thread_manager::ThreadManagerState;
 use codex_features::Feature;
 use codex_login::CodexAuth;
+use codex_protocol::AgentPath;
 use codex_protocol::ThreadId;
 use codex_protocol::error::CodexErrorDetails;
 use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::InterAgentCommunication;
+use codex_protocol::protocol::Op;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::SubAgentSource;
+use codex_protocol::protocol::ThreadSettingsOverrides;
 use codex_protocol::protocol::ThreadSource;
 use codex_protocol::protocol::TurnAbortReason;
 use codex_protocol::protocol::TurnAbortedEvent;
@@ -58,6 +62,29 @@ async fn residency_slot_reservation_unloads_oldest_idle_v2_agent() {
     first_slot.commit(first.thread_id);
     mark_thread_completed(first.thread.as_ref()).await;
 
+    // Public message submission must honor the same eviction gate as MAv2 sends.
+    let eviction_gate = Arc::clone(&first.thread.residency_gate).write_owned().await;
+    let communication = InterAgentCommunication::new(
+        AgentPath::root(),
+        AgentPath::try_from("/root/worker_1").unwrap(),
+        Vec::new(),
+        "queued notice".into(),
+        /*trigger_turn*/ false,
+    );
+    let delivery = first.thread.submit(Op::InterAgentCommunication {
+        communication: communication.clone(),
+        start_options: Default::default(),
+    });
+    tokio::pin!(delivery);
+    assert!(futures::poll!(delivery.as_mut()).is_pending());
+    drop(eviction_gate);
+    delivery.await.expect("submit after eviction gate releases");
+    first
+        .thread
+        .update_thread_settings(ThreadSettingsOverrides::default())
+        .await
+        .expect("dispatch queued notice before eviction");
+
     let second_slot = control
         .reserve_v2_residency_slot(
             &state,
@@ -74,6 +101,14 @@ async fn residency_slot_reservation_unloads_oldest_idle_v2_agent() {
         },
         Ok(_) => panic!("expected evicted thread to be missing"),
     }
+    let retained = first
+        .thread
+        .session
+        .services
+        .local_agent_runtime
+        .mailboxes
+        .take(first.thread_id);
+    assert_eq!(retained, vec![communication.into()]);
     let second = spawn_v2_subagent(&control, &state, config, root.thread_id, "worker-2").await;
     second_slot.commit(second.thread_id);
 

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

@@ -264,7 +264,7 @@ impl LocalAgentControl {
         if owner_thread_id.is_none()
             && let Ok(thread) = state.get_thread(thread_id).await
         {
-            return self.runtime.pin_v2_residency(&state, &thread).await;
+            return self.runtime.pin_v2_residency(&thread).await;
         }
         if self
             .runtime
@@ -328,7 +328,7 @@ impl LocalAgentControl {
             }
             if let Ok(thread) = state.get_thread(thread_id).await {
                 self.validate_loaded_v2_child(&thread, parent_thread_id)?;
-                return self.runtime.pin_v2_residency(&state, &thread).await;
+                return self.runtime.pin_v2_residency(&thread).await;
             }
         }
         let parent = if let Some(parent) = parent {
@@ -589,9 +589,7 @@ impl LocalAgentControl {
                 // Finish registration even if the sender is cancelled while acquiring its pin.
                 tokio::spawn(async move {
                     let _membership = membership;
-                    let residency_pin = runtime
-                        .pin_v2_residency(&state, &reloaded_thread.thread)
-                        .await?;
+                    let residency_pin = runtime.pin_v2_residency(&reloaded_thread.thread).await?;
                     runtime.registry.clear_evicted_environments(thread_id);
                     residency_slot.commit(reloaded_thread.thread_id);
                     // Register before listeners can forward events from the resumed thread.
@@ -616,7 +614,7 @@ impl LocalAgentControl {
                     }
                     self.runtime.registry.clear_evicted_environments(thread_id);
                     drop(residency_slot);
-                    return self.runtime.pin_v2_residency(&state, &thread).await;
+                    return self.runtime.pin_v2_residency(&thread).await;
                 }
                 Err(err)
             }

codex-rs/core/src/agent_message_board.rs

@@ -2,7 +2,7 @@
 //!
 //! The extension owns backend construction, storage and tool registration. This adapter
 //! selects configuration and supplies host capabilities. It never starts or restores
-//! recipients and never queues a notification for an idle agent.
+//! recipients during lookup. Notifications use ordinary agent-message scheduling.
 
 use crate::CodexThread;
 use crate::ThreadManager;
@@ -28,6 +28,7 @@ use codex_protocol::error::CodexErr;
 use codex_protocol::error::CodexErrorDetails;
 use codex_protocol::error::Result;
 use codex_protocol::protocol::InterAgentCommunication;
+use codex_protocol::protocol::Op;
 use futures::future::BoxFuture;
 use std::sync::Arc;
 use std::sync::Weak;
@@ -44,7 +45,7 @@ pub fn install_agent_message_board(
         Some((
             board,
             Arc::new(LocalBoardHost {
-                expected_turn_id: Some(input.turn_id.into()),
+                remote_notifications: true,
                 ..host.as_ref().clone()
             }),
         ))
@@ -91,7 +92,7 @@ pub fn install_agent_message_board(
                     manager: manager.clone(),
                     tree,
                     caller,
-                    expected_turn_id: None,
+                    remote_notifications: false,
                     connection,
                 },
             ))
@@ -104,7 +105,7 @@ struct LocalBoardHost {
     manager: Weak<ThreadManager>,
     tree: SessionId,
     caller: ThreadId,
-    expected_turn_id: Option<String>,
+    remote_notifications: bool,
     connection: Option<Arc<LocalBoardConnection>>,
 }
 
@@ -237,7 +238,7 @@ impl MessageBoardHost for LocalBoardHost {
             );
             // Remote metadata is untrusted. Skip an oversized notice without closing
             // the receiver; the post remains available through the board tools.
-            if self.expected_turn_id.is_some()
+            if self.remote_notifications
                 && communication.content.len()
                     + communication.author.as_str().len()
                     + communication.recipient.as_str().len()
@@ -247,22 +248,13 @@ impl MessageBoardHost for LocalBoardHost {
                     "remote board notice exceeds the context budget".into(),
                 ));
             }
-            Ok(
-                if recipient
-                    .session
-                    .input_queue
-                    .deliver_mailbox_communication_to_current_turn(
-                        &recipient.session.active_turn,
-                        communication,
-                        self.expected_turn_id.as_deref(),
-                    )
-                    .await
-                {
-                    NotificationDelivery::Accepted
-                } else {
-                    NotificationDelivery::SkippedInactive
-                },
-            )
+            recipient
+                .submit(Op::InterAgentCommunication {
+                    communication,
+                    start_options: Default::default(),
+                })
+                .await?;
+            Ok(NotificationDelivery::Accepted)
         })
     }
 }

codex-rs/core/src/codex_thread.rs

@@ -242,7 +242,7 @@ impl CodexThread {
     }
 
     pub async fn submit(&self, op: Op) -> CodexResult<String> {
-        self.io.submit(op).await
+        self.submit_with_trace(op, /*trace*/ None).await
     }
 
     /// Returns the session telemetry handle for thread-scoped production instrumentation.
@@ -453,10 +453,24 @@ impl CodexThread {
         op: Op,
         trace: Option<W3cTraceContext>,
     ) -> CodexResult<String> {
+        // Use the same admission guard as MAv2 sends, retaining it through dispatch.
+        let residency_guard = if matches!(op, Op::InterAgentCommunication { .. }) {
+            self.session
+                .services
+                .local_agent_runtime
+                .pin_v2_residency(self)
+                .await?
+        } else {
+            None
+        };
         self.io
             .submit_with_trace(
-                op, trace, /*parent_turn_id*/ None, /*root_turn_id*/ None,
-                /*residency_guard*/ None, /*message_delivery*/ None,
+                op,
+                trace,
+                /*parent_turn_id*/ None,
+                /*root_turn_id*/ None,
+                residency_guard,
+                /*message_delivery*/ None,
             )
             .await
     }

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

@@ -7,7 +7,6 @@ use codex_diagnostics::GaugeGuard;
 use codex_history::ResponseItemEnvelope;
 use codex_protocol::ThreadId;
 use codex_protocol::models::ResponseItem;
-use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::turn_input::TurnStartOptions;
 use codex_protocol::user_input::UserInput;
 use serde::Deserialize;
@@ -202,38 +201,6 @@ impl InputQueue {
         (activity_rx, pending_activity)
     }
 
-    #[expect(
-        clippy::await_holding_invalid_type,
-        reason = "active turn checks and turn state updates must remain atomic"
-    )]
-    pub(crate) async fn deliver_mailbox_communication_to_current_turn(
-        &self,
-        active_turn: &Mutex<Option<ActiveTurn>>,
-        communication: InterAgentCommunication,
-        expected_turn_id: Option<&str>,
-    ) -> bool {
-        let active = active_turn.lock().await;
-        let Some(active_turn) = active.as_ref().filter(|turn| {
-            turn.task.as_ref().is_some_and(|task| {
-                expected_turn_id.is_none_or(|id| {
-                    task.turn_context.sub_id == id && !task.cancellation_token.is_cancelled()
-                })
-            })
-        }) else {
-            return false;
-        };
-        let mut turn_state = active_turn.turn_state.lock().await;
-        if !turn_state.accepts_mailbox_delivery_for_current_turn() {
-            return false;
-        }
-        turn_state
-            .pending_input
-            .items
-            .push(TurnInput::InterAgentCommunication(communication.into()));
-        self.activity_tx.send_replace(InputQueueActivity::Mailbox);
-        true
-    }
-
     pub(crate) async fn enqueue_mailbox_communication(
         &self,
         communication: crate::PendingAgentMessage,
@@ -556,6 +523,7 @@ mod tests {
     use super::*;
     use codex_history::CodexHarnessMetadata;
     use codex_protocol::AgentPath;
+    use codex_protocol::protocol::InterAgentCommunication;
     use codex_protocol::user_input::UserInput;
     use pretty_assertions::assert_eq;
 

codex-rs/core/src/thread_manager.rs

@@ -1886,7 +1886,7 @@ impl ThreadManagerState {
                 .session
                 .services
                 .local_agent_runtime
-                .pin_v2_residency(self, &thread)
+                .pin_v2_residency(&thread)
                 .await?
         } else {
             None