[codex] Move pending input into input queue (#22728)

## Why Pending model input was split across `Session`, `TurnState`, and the agent mailbox. That made it easy for new paths to manage queued user input or mailbox delivery outside the intended ownership boundary. This PR consolidates the model-facing input lifecycle behind the session input queue so turn-local pending input, next-turn queued items, and mailbox delivery coordination are owned in one place. ## What Changed - Added `session/input_queue.rs` to own pending input queues and mailbox delivery coordination. - Removed the standalone `agent/mailbox.rs` channel wrapper and store mailbox items directly in the input queue. - Moved pending-input mutations off `TurnState`; `TurnState` now exposes the queue-owned storage directly for now. - Routed abort cleanup, mailbox delivery phase changes, next-turn queued items, and active-turn pending input through `InputQueue`. - Boxed stack-heavy agent resume/fork startup futures that the refactor pushed over the default test stack. - Updated session, task, goal, stream-event, and multi-agent call sites and tests to use the new queue ownership. ## Verification - `cargo test -p codex-core --lib agent::control::tests` - `cargo test -p codex-core --lib agent::control::tests::resume_closed_child_reopens_open_descendants -- --exact` - `cargo test -p codex-core --lib agent::control::tests::spawn_agent_fork_last_n_turns_keeps_only_recent_turns -- --exact` - `cargo test -p codex-core --lib agent::control::tests::resume_thread_subagent_restores_stored_nickname_and_role -- --exact` - `cargo test -p codex-core` was also run; it completed with 1814 passed, 4 ignored, and one timeout in `agent::control::tests::resume_thread_subagent_restores_stored_nickname_and_role`, which passed when rerun in isolation.
2026-07-01 00:31:56 +08:00 · 2026-05-18 15:43:01 -07:00
parent a66e0e9c4b
commit afa0101ae2
18 changed files with 623 additions and 524 deletions
@@ -319,7 +319,9 @@ pub async fn inter_agent_communication(
    communication: InterAgentCommunication,
 ) {
    let trigger_turn = communication.trigger_turn;
-    sess.enqueue_mailbox_communication(communication);
+    sess.input_queue
+        .enqueue_mailbox_communication(communication)
+        .await;
    if trigger_turn {
        sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id)
            .await;
@@ -961,7 +963,9 @@ Approved action:
    }];

    if let Err(items) = sess.inject_response_items(items).await {
-        sess.queue_response_items_for_next_turn(items).await;
+        sess.input_queue
+            .queue_response_items_for_next_turn(items)
+            .await;
    }
 }

@@ -0,0 +1,392 @@
+use crate::state::ActiveTurn;
+use crate::state::MailboxDeliveryPhase;
+use crate::state::TurnState;
+use codex_protocol::models::ResponseInputItem;
+use codex_protocol::protocol::InterAgentCommunication;
+use std::collections::VecDeque;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tokio::sync::watch;
+
+/// Turn-local pending input storage owned by the input queue flow.
+#[derive(Default)]
+pub(crate) struct TurnInputQueue {
+    items: Vec<ResponseInputItem>,
+}
+
+/// Session-scoped pending input storage and active-turn mailbox delivery coordination.
+pub(crate) struct InputQueue {
+    mailbox_tx: watch::Sender<()>,
+    mailbox_pending_mails: Mutex<VecDeque<InterAgentCommunication>>,
+
+    idle_pending_input: Mutex<Vec<ResponseInputItem>>,
+}
+
+impl InputQueue {
+    pub(crate) fn new() -> Self {
+        let (mailbox_tx, _) = watch::channel(());
+        Self {
+            mailbox_tx,
+            mailbox_pending_mails: Mutex::new(VecDeque::new()),
+            idle_pending_input: Mutex::new(Vec::new()),
+        }
+    }
+
+    pub(crate) async fn subscribe_mailbox(&self) -> watch::Receiver<()> {
+        let mut mailbox_rx = self.mailbox_tx.subscribe();
+        if self.has_pending_mailbox_items().await {
+            mailbox_rx.mark_changed();
+        }
+        mailbox_rx
+    }
+
+    pub(crate) async fn enqueue_mailbox_communication(
+        &self,
+        communication: InterAgentCommunication,
+    ) {
+        self.mailbox_pending_mails
+            .lock()
+            .await
+            .push_back(communication);
+        self.mailbox_tx.send_replace(());
+    }
+
+    pub(crate) async fn has_pending_mailbox_items(&self) -> bool {
+        !self.mailbox_pending_mails.lock().await.is_empty()
+    }
+
+    pub(crate) async fn has_trigger_turn_mailbox_items(&self) -> bool {
+        self.mailbox_pending_mails
+            .lock()
+            .await
+            .iter()
+            .any(|mail| mail.trigger_turn)
+    }
+
+    pub(crate) async fn drain_mailbox_input_items(&self) -> Vec<ResponseInputItem> {
+        self.mailbox_pending_mails
+            .lock()
+            .await
+            .drain(..)
+            .map(|mail| mail.to_response_input_item())
+            .collect()
+    }
+
+    pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec<ResponseInputItem>) {
+        if items.is_empty() {
+            return;
+        }
+
+        self.idle_pending_input.lock().await.extend(items);
+    }
+
+    pub(crate) async fn take_queued_response_items_for_next_turn(&self) -> Vec<ResponseInputItem> {
+        std::mem::take(&mut *self.idle_pending_input.lock().await)
+    }
+
+    pub(crate) async fn has_queued_response_items_for_next_turn(&self) -> bool {
+        !self.idle_pending_input.lock().await.is_empty()
+    }
+
+    pub(crate) async fn turn_state_for_sub_id(
+        &self,
+        active_turn: &Mutex<Option<ActiveTurn>>,
+        sub_id: &str,
+    ) -> Option<Arc<Mutex<TurnState>>> {
+        let active = active_turn.lock().await;
+        active.as_ref().and_then(|active_turn| {
+            active_turn
+                .tasks
+                .contains_key(sub_id)
+                .then(|| Arc::clone(&active_turn.turn_state))
+        })
+    }
+
+    /// Clear any pending waiters and input buffered for the current turn.
+    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.items.clear();
+    }
+
+    pub(crate) async fn defer_mailbox_delivery_to_next_turn(
+        &self,
+        active_turn: &Mutex<Option<ActiveTurn>>,
+        sub_id: &str,
+    ) {
+        let turn_state = self.turn_state_for_sub_id(active_turn, sub_id).await;
+        let Some(turn_state) = turn_state else {
+            return;
+        };
+        let mut turn_state = turn_state.lock().await;
+        if !turn_state.pending_input.items.is_empty() {
+            return;
+        }
+        turn_state.set_mailbox_delivery_phase(MailboxDeliveryPhase::NextTurn);
+    }
+
+    pub(crate) async fn accept_mailbox_delivery_for_current_turn(
+        &self,
+        active_turn: &Mutex<Option<ActiveTurn>>,
+        sub_id: &str,
+    ) {
+        let turn_state = self.turn_state_for_sub_id(active_turn, sub_id).await;
+        let Some(turn_state) = turn_state else {
+            return;
+        };
+        self.accept_mailbox_delivery_for_turn_state(turn_state.as_ref())
+            .await;
+    }
+
+    pub(super) async fn accept_mailbox_delivery_for_turn_state(
+        &self,
+        turn_state: &Mutex<TurnState>,
+    ) {
+        turn_state
+            .lock()
+            .await
+            .accept_mailbox_delivery_for_current_turn();
+    }
+
+    pub(super) async fn push_pending_input_and_accept_mailbox_delivery_for_turn_state(
+        &self,
+        turn_state: &Mutex<TurnState>,
+        input: ResponseInputItem,
+    ) {
+        let mut turn_state = turn_state.lock().await;
+        turn_state.pending_input.items.push(input);
+        turn_state.accept_mailbox_delivery_for_current_turn();
+    }
+
+    pub(crate) async fn extend_pending_input_for_turn_state(
+        &self,
+        turn_state: &Mutex<TurnState>,
+        input: Vec<ResponseInputItem>,
+    ) {
+        turn_state.lock().await.pending_input.items.extend(input);
+    }
+
+    pub(crate) async fn take_pending_input_for_turn_state(
+        &self,
+        turn_state: &Mutex<TurnState>,
+    ) -> Vec<ResponseInputItem> {
+        turn_state.lock().await.pending_input.items.split_off(0)
+    }
+
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "active turn checks and turn state updates must remain atomic"
+    )]
+    pub(crate) async fn inject_response_items(
+        &self,
+        active_turn: &Mutex<Option<ActiveTurn>>,
+        input: Vec<ResponseInputItem>,
+    ) -> Result<(), Vec<ResponseInputItem>> {
+        let mut active = active_turn.lock().await;
+        match active.as_mut() {
+            Some(active_turn) => {
+                active_turn
+                    .turn_state
+                    .lock()
+                    .await
+                    .pending_input
+                    .items
+                    .extend(input);
+                Ok(())
+            }
+            None => Err(input),
+        }
+    }
+
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "active turn checks and turn state updates must remain atomic"
+    )]
+    pub(crate) async fn prepend_pending_input(
+        &self,
+        active_turn: &Mutex<Option<ActiveTurn>>,
+        mut input: Vec<ResponseInputItem>,
+    ) -> Result<(), ()> {
+        let mut active = active_turn.lock().await;
+        match active.as_mut() {
+            Some(active_turn) => {
+                let mut turn_state = active_turn.turn_state.lock().await;
+                if !input.is_empty() {
+                    let pending_input = &mut turn_state.pending_input;
+                    input.append(&mut pending_input.items);
+                    pending_input.items = input;
+                }
+                Ok(())
+            }
+            None => Err(()),
+        }
+    }
+
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "active turn checks and turn state updates must remain atomic"
+    )]
+    pub(crate) async fn get_pending_input(
+        &self,
+        active_turn: &Mutex<Option<ActiveTurn>>,
+    ) -> Vec<ResponseInputItem> {
+        let (pending_input, accepts_mailbox_delivery) = {
+            let mut active = active_turn.lock().await;
+            match active.as_mut() {
+                Some(active_turn) => {
+                    let mut turn_state = active_turn.turn_state.lock().await;
+                    (
+                        turn_state.pending_input.items.split_off(0),
+                        turn_state.accepts_mailbox_delivery_for_current_turn(),
+                    )
+                }
+                None => (Vec::new(), true),
+            }
+        };
+        if !accepts_mailbox_delivery {
+            return pending_input;
+        }
+        let mailbox_items = self.drain_mailbox_input_items().await;
+        if pending_input.is_empty() {
+            mailbox_items
+        } else if mailbox_items.is_empty() {
+            pending_input
+        } else {
+            let mut pending_input = pending_input;
+            pending_input.extend(mailbox_items);
+            pending_input
+        }
+    }
+
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "active turn checks and turn state reads must remain atomic"
+    )]
+    pub(crate) async fn has_pending_input(&self, active_turn: &Mutex<Option<ActiveTurn>>) -> bool {
+        let (has_turn_pending_input, accepts_mailbox_delivery) = {
+            let active = active_turn.lock().await;
+            match active.as_ref() {
+                Some(active_turn) => {
+                    let turn_state = active_turn.turn_state.lock().await;
+                    (
+                        !turn_state.pending_input.items.is_empty(),
+                        turn_state.accepts_mailbox_delivery_for_current_turn(),
+                    )
+                }
+                None => (false, true),
+            }
+        };
+        if has_turn_pending_input {
+            return true;
+        }
+        if !accepts_mailbox_delivery {
+            return false;
+        }
+        self.has_pending_mailbox_items().await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use codex_protocol::AgentPath;
+    use pretty_assertions::assert_eq;
+
+    fn make_mail(
+        author: AgentPath,
+        recipient: AgentPath,
+        content: &str,
+        trigger_turn: bool,
+    ) -> InterAgentCommunication {
+        InterAgentCommunication::new(
+            author,
+            recipient,
+            Vec::new(),
+            content.to_string(),
+            trigger_turn,
+        )
+    }
+
+    #[tokio::test]
+    async fn input_queue_notifies_mailbox_subscribers() {
+        let input_queue = InputQueue::new();
+        let mut mailbox_rx = input_queue.subscribe_mailbox().await;
+
+        input_queue
+            .enqueue_mailbox_communication(make_mail(
+                AgentPath::root(),
+                AgentPath::try_from("/root/worker").expect("agent path"),
+                "one",
+                /*trigger_turn*/ false,
+            ))
+            .await;
+        input_queue
+            .enqueue_mailbox_communication(make_mail(
+                AgentPath::root(),
+                AgentPath::try_from("/root/worker").expect("agent path"),
+                "two",
+                /*trigger_turn*/ false,
+            ))
+            .await;
+
+        mailbox_rx.changed().await.expect("mailbox update");
+    }
+
+    #[tokio::test]
+    async fn input_queue_drains_mailbox_in_delivery_order() {
+        let input_queue = InputQueue::new();
+        let mail_one = make_mail(
+            AgentPath::root(),
+            AgentPath::try_from("/root/worker").expect("agent path"),
+            "one",
+            /*trigger_turn*/ false,
+        );
+        let mail_two = make_mail(
+            AgentPath::try_from("/root/worker").expect("agent path"),
+            AgentPath::root(),
+            "two",
+            /*trigger_turn*/ false,
+        );
+
+        input_queue
+            .enqueue_mailbox_communication(mail_one.clone())
+            .await;
+        input_queue
+            .enqueue_mailbox_communication(mail_two.clone())
+            .await;
+
+        assert_eq!(
+            input_queue.drain_mailbox_input_items().await,
+            vec![
+                mail_one.to_response_input_item(),
+                mail_two.to_response_input_item()
+            ]
+        );
+        assert!(!input_queue.has_pending_mailbox_items().await);
+    }
+
+    #[tokio::test]
+    async fn input_queue_tracks_pending_trigger_turn_mail() {
+        let input_queue = InputQueue::new();
+
+        input_queue
+            .enqueue_mailbox_communication(make_mail(
+                AgentPath::root(),
+                AgentPath::try_from("/root/worker").expect("agent path"),
+                "queued",
+                /*trigger_turn*/ false,
+            ))
+            .await;
+        assert!(!input_queue.has_trigger_turn_mailbox_items().await);
+
+        input_queue
+            .enqueue_mailbox_communication(make_mail(
+                AgentPath::root(),
+                AgentPath::try_from("/root/worker").expect("agent path"),
+                "wake",
+                /*trigger_turn*/ true,
+            ))
+            .await;
+        assert!(input_queue.has_trigger_turn_mailbox_items().await);
+    }
+}
@@ -10,8 +10,6 @@ use std::time::UNIX_EPOCH;

 use crate::agent::AgentControl;
 use crate::agent::AgentStatus;
-use crate::agent::Mailbox;
-use crate::agent::MailboxReceiver;
 use crate::agent::agent_status_from_event;
 use crate::agent::status::is_final;
 use crate::attestation::AttestationProvider;
@@ -193,6 +191,7 @@ use codex_protocol::exec_output::StreamOutput;

 mod config_lock;
 mod handlers;
+mod input_queue;
 mod mcp;
 mod multi_agents;
 mod review;
@@ -206,6 +205,7 @@ use self::config_lock::validate_config_lock_if_configured;
 #[cfg(test)]
 use self::handlers::submission_dispatch_span;
 use self::handlers::submission_loop;
+pub(crate) use self::input_queue::TurnInputQueue;
 use self::review::spawn_review_thread;
 use self::session::AppServerClientMetadata;
 use self::session::Session;
@@ -289,8 +289,6 @@ use crate::rollout::map_session_init_error;
 use crate::session_startup_prewarm::SessionStartupPrewarmHandle;
 use crate::shell;
 use crate::shell_snapshot::ShellSnapshot;
-use crate::state::ActiveTurn;
-use crate::state::MailboxDeliveryPhase;
 use crate::state::PendingRequestPermissions;
 use crate::state::SessionServices;
 use crate::state::SessionState;
@@ -3163,195 +3161,36 @@ impl Session {
                .set_responsesapi_client_metadata(responsesapi_client_metadata);
        }

-        let mut turn_state = active_turn.turn_state.lock().await;
-        turn_state.push_pending_input(input.into());
-        turn_state.accept_mailbox_delivery_for_current_turn();
+        self.input_queue
+            .push_pending_input_and_accept_mailbox_delivery_for_turn_state(
+                active_turn.turn_state.as_ref(),
+                input.into(),
+            )
+            .await;
        Ok(active_turn_id.clone())
    }

    /// Returns the input if there was no task running to inject into.
-    #[expect(
-        clippy::await_holding_invalid_type,
-        reason = "active turn checks and turn state updates must remain atomic"
-    )]
    pub async fn inject_response_items(
        &self,
        input: Vec<ResponseInputItem>,
    ) -> Result<(), Vec<ResponseInputItem>> {
-        let mut active = self.active_turn.lock().await;
-        match active.as_mut() {
-            Some(at) => {
-                let mut ts = at.turn_state.lock().await;
-                for item in input {
-                    ts.push_pending_input(item);
-                }
-                Ok(())
-            }
-            None => Err(input),
-        }
-    }
-
-    pub(crate) async fn defer_mailbox_delivery_to_next_turn(&self, sub_id: &str) {
-        let turn_state = self.turn_state_for_sub_id(sub_id).await;
-        let Some(turn_state) = turn_state else {
-            return;
-        };
-        let mut turn_state = turn_state.lock().await;
-        if turn_state.has_pending_input() {
-            return;
-        }
-        turn_state.set_mailbox_delivery_phase(MailboxDeliveryPhase::NextTurn);
-    }
-
-    pub(crate) async fn accept_mailbox_delivery_for_current_turn(&self, sub_id: &str) {
-        let turn_state = self.turn_state_for_sub_id(sub_id).await;
-        let Some(turn_state) = turn_state else {
-            return;
-        };
-        turn_state
-            .lock()
+        self.input_queue
+            .inject_response_items(&self.active_turn, input)
            .await
-            .set_mailbox_delivery_phase(MailboxDeliveryPhase::CurrentTurn);
    }

    pub(crate) async fn record_memory_citation_for_turn(&self, sub_id: &str) {
-        let turn_state = self.turn_state_for_sub_id(sub_id).await;
+        let turn_state = self
+            .input_queue
+            .turn_state_for_sub_id(&self.active_turn, sub_id)
+            .await;
        let Some(turn_state) = turn_state else {
            return;
        };
        turn_state.lock().await.has_memory_citation = true;
    }

-    async fn turn_state_for_sub_id(
-        &self,
-        sub_id: &str,
-    ) -> Option<Arc<tokio::sync::Mutex<crate::state::TurnState>>> {
-        let active = self.active_turn.lock().await;
-        active.as_ref().and_then(|active_turn| {
-            active_turn
-                .tasks
-                .contains_key(sub_id)
-                .then(|| Arc::clone(&active_turn.turn_state))
-        })
-    }
-
-    pub(crate) fn subscribe_mailbox_seq(&self) -> watch::Receiver<u64> {
-        self.mailbox.subscribe()
-    }
-
-    pub(crate) fn enqueue_mailbox_communication(&self, communication: InterAgentCommunication) {
-        self.mailbox.send(communication);
-    }
-
-    pub(crate) async fn has_trigger_turn_mailbox_items(&self) -> bool {
-        self.mailbox_rx.lock().await.has_pending_trigger_turn()
-    }
-
-    pub(crate) async fn has_pending_mailbox_items(&self) -> bool {
-        self.mailbox_rx.lock().await.has_pending()
-    }
-
-    #[expect(
-        clippy::await_holding_invalid_type,
-        reason = "active turn checks and turn state updates must remain atomic"
-    )]
-    pub async fn prepend_pending_input(&self, input: Vec<ResponseInputItem>) -> Result<(), ()> {
-        let mut active = self.active_turn.lock().await;
-        match active.as_mut() {
-            Some(at) => {
-                let mut ts = at.turn_state.lock().await;
-                ts.prepend_pending_input(input);
-                Ok(())
-            }
-            None => Err(()),
-        }
-    }
-
-    #[expect(
-        clippy::await_holding_invalid_type,
-        reason = "active turn checks and turn state updates must remain atomic"
-    )]
-    pub async fn get_pending_input(&self) -> Vec<ResponseInputItem> {
-        let (pending_input, accepts_mailbox_delivery) = {
-            let mut active = self.active_turn.lock().await;
-            match active.as_mut() {
-                Some(at) => {
-                    let mut ts = at.turn_state.lock().await;
-                    (
-                        ts.take_pending_input(),
-                        ts.accepts_mailbox_delivery_for_current_turn(),
-                    )
-                }
-                None => (Vec::new(), true),
-            }
-        };
-        if !accepts_mailbox_delivery {
-            return pending_input;
-        }
-        let mailbox_items = {
-            let mut mailbox_rx = self.mailbox_rx.lock().await;
-            mailbox_rx
-                .drain()
-                .into_iter()
-                .map(|mail| mail.to_response_input_item())
-                .collect::<Vec<_>>()
-        };
-        if pending_input.is_empty() {
-            mailbox_items
-        } else if mailbox_items.is_empty() {
-            pending_input
-        } else {
-            let mut pending_input = pending_input;
-            pending_input.extend(mailbox_items);
-            pending_input
-        }
-    }
-
-    /// Queue response items to be injected into the next active turn created for this session.
-    pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec<ResponseInputItem>) {
-        if items.is_empty() {
-            return;
-        }
-
-        let mut idle_pending_input = self.idle_pending_input.lock().await;
-        idle_pending_input.extend(items);
-    }
-
-    pub(crate) async fn take_queued_response_items_for_next_turn(&self) -> Vec<ResponseInputItem> {
-        std::mem::take(&mut *self.idle_pending_input.lock().await)
-    }
-
-    pub(crate) async fn has_queued_response_items_for_next_turn(&self) -> bool {
-        !self.idle_pending_input.lock().await.is_empty()
-    }
-
-    #[expect(
-        clippy::await_holding_invalid_type,
-        reason = "active turn checks and turn state reads must remain atomic"
-    )]
-    pub async fn has_pending_input(&self) -> bool {
-        let (has_turn_pending_input, accepts_mailbox_delivery) = {
-            let active = self.active_turn.lock().await;
-            match active.as_ref() {
-                Some(at) => {
-                    let ts = at.turn_state.lock().await;
-                    (
-                        ts.has_pending_input(),
-                        ts.accepts_mailbox_delivery_for_current_turn(),
-                    )
-                }
-                None => (false, true),
-            }
-        };
-        if has_turn_pending_input {
-            return true;
-        }
-        if !accepts_mailbox_delivery {
-            return false;
-        }
-        self.has_pending_mailbox_items().await
-    }
-
    pub async fn interrupt_task(self: &Arc<Self>) {
        info!("interrupt received: abort current task, if any");
        let had_active_turn = self.active_turn.lock().await.is_some();
@@ -1,5 +1,7 @@
+use super::input_queue::InputQueue;
 use super::*;
 use crate::goals::GoalRuntimeState;
+use crate::state::ActiveTurn;
 use codex_protocol::SessionId;
 use codex_protocol::config_types::ServiceTier;
 use codex_protocol::permissions::FileSystemPath;
@@ -27,9 +29,7 @@ pub(crate) struct Session {
    pub(super) pending_mcp_server_refresh_config: Mutex<Option<McpServerRefreshConfig>>,
    pub(crate) conversation: Arc<RealtimeConversationManager>,
    pub(crate) active_turn: Mutex<Option<ActiveTurn>>,
-    pub(super) mailbox: Mailbox,
-    pub(super) mailbox_rx: Mutex<MailboxReceiver>,
-    pub(super) idle_pending_input: Mutex<Vec<ResponseInputItem>>, // TODO (jif) merge with mailbox!
+    pub(crate) input_queue: InputQueue,
    pub(crate) goal_runtime: GoalRuntimeState,
    pub(crate) guardian_review_session: GuardianReviewSessionManager,
    pub(crate) services: SessionServices,
@@ -950,7 +950,6 @@ impl Session {
            let (out_of_band_elicitation_paused, _out_of_band_elicitation_paused_rx) =
                watch::channel(false);

-            let (mailbox, mailbox_rx) = Mailbox::new();
            let sess = Arc::new(Session {
                conversation_id: thread_id,
                installation_id,
@@ -963,9 +962,7 @@ impl Session {
                pending_mcp_server_refresh_config: Mutex::new(None),
                conversation: Arc::new(RealtimeConversationManager::new()),
                active_turn: Mutex::new(None),
-                mailbox,
-                mailbox_rx: Mutex::new(mailbox_rx),
-                idle_pending_input: Mutex::new(Vec::new()),
+                input_queue: InputQueue::new(),
                goal_runtime: GoalRuntimeState::new(),
                guardian_review_session: GuardianReviewSessionManager::default(),
                services,
@@ -1130,7 +1127,7 @@ impl Session {
            };

            // record_initial_history can emit events. We record only after the SessionConfiguredEvent is emitted.
-            sess.record_initial_history(initial_history).await;
+            Box::pin(sess.record_initial_history(initial_history)).await;
            {
                let mut state = sess.state.lock().await;
                state.set_pending_session_start_source(Some(session_start_source));
@@ -4394,7 +4394,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
        /*goal_tools_supported*/ true,
    );

-    let (mailbox, mailbox_rx) = crate::agent::Mailbox::new();
    let session = Session {
        conversation_id: thread_id,
        installation_id: "11111111-1111-4111-8111-111111111111".to_string(),
@@ -4407,9 +4406,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
        pending_mcp_server_refresh_config: Mutex::new(None),
        conversation: Arc::new(RealtimeConversationManager::new()),
        active_turn: Mutex::new(None),
-        mailbox,
-        mailbox_rx: Mutex::new(mailbox_rx),
-        idle_pending_input: Mutex::new(Vec::new()),
+        input_queue: super::input_queue::InputQueue::new(),
        goal_runtime: crate::goals::GoalRuntimeState::new(),
        guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
        services,
@@ -6252,7 +6249,6 @@ where
        /*goal_tools_supported*/ true,
    ));

-    let (mailbox, mailbox_rx) = crate::agent::Mailbox::new();
    let session = Arc::new(Session {
        conversation_id: thread_id,
        installation_id: "11111111-1111-4111-8111-111111111111".to_string(),
@@ -6265,9 +6261,7 @@ where
        pending_mcp_server_refresh_config: Mutex::new(None),
        conversation: Arc::new(RealtimeConversationManager::new()),
        active_turn: Mutex::new(None),
-        mailbox,
-        mailbox_rx: Mutex::new(mailbox_rx),
-        idle_pending_input: Mutex::new(Vec::new()),
+        input_queue: super::input_queue::InputQueue::new(),
        goal_runtime: crate::goals::GoalRuntimeState::new(),
        guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
        services,
@@ -8098,7 +8092,7 @@ async fn steer_input_returns_active_turn_id() {
        .expect("steering with matching expected turn id should succeed");

    assert_eq!(turn_id, tc.sub_id);
-    assert!(sess.has_pending_input().await);
+    assert!(sess.input_queue.has_pending_input(&sess.active_turn).await);
 }

 #[tokio::test]
@@ -8144,7 +8138,7 @@ async fn prepend_pending_input_keeps_older_tail_ahead_of_newer_input() {
        .await
        .expect("inject initial pending input into active turn");

-    let drained = sess.get_pending_input().await;
+    let drained = sess.input_queue.get_pending_input(&sess.active_turn).await;
    assert_eq!(drained, vec![blocked, later.clone()]);

    sess.inject_response_items(vec![newer.clone()])
@@ -8153,11 +8147,15 @@ async fn prepend_pending_input_keeps_older_tail_ahead_of_newer_input() {

    let mut drained_iter = drained.into_iter();
    let _blocked = drained_iter.next().expect("blocked prompt should exist");
-    sess.prepend_pending_input(drained_iter.collect())
+    sess.input_queue
+        .prepend_pending_input(&sess.active_turn, drained_iter.collect())
        .await
        .expect("requeue later pending input at the front of the queue");

-    assert_eq!(sess.get_pending_input().await, vec![later, newer]);
+    assert_eq!(
+        sess.input_queue.get_pending_input(&sess.active_turn).await,
+        vec![later, newer]
+    );
 }

 #[tokio::test]
@@ -8171,7 +8169,8 @@ async fn queued_response_items_for_next_turn_move_into_next_active_turn() {
        phase: None,
    };

-    sess.queue_response_items_for_next_turn(vec![queued_item.clone()])
+    sess.input_queue
+        .queue_response_items_for_next_turn(vec![queued_item.clone()])
        .await;

    sess.spawn_task(
@@ -8184,7 +8183,10 @@ async fn queued_response_items_for_next_turn_move_into_next_active_turn() {
    )
    .await;

-    assert_eq!(sess.get_pending_input().await, vec![queued_item]);
+    assert_eq!(
+        sess.input_queue.get_pending_input(&sess.active_turn).await,
+        vec![queued_item]
+    );
 }

 #[tokio::test]
@@ -8198,13 +8200,18 @@ async fn idle_interrupt_does_not_wake_queued_next_turn_items() {
        phase: None,
    };

-    sess.queue_response_items_for_next_turn(vec![queued_item])
+    sess.input_queue
+        .queue_response_items_for_next_turn(vec![queued_item])
        .await;

    sess.abort_all_tasks(TurnAbortReason::Interrupted).await;

    assert!(sess.active_turn.lock().await.is_none());
-    assert!(sess.has_queued_response_items_for_next_turn().await);
+    assert!(
+        sess.input_queue
+            .has_queued_response_items_for_next_turn()
+            .await
+    );
 }

 #[tokio::test]
@@ -8222,16 +8229,17 @@ async fn abort_empty_active_turn_preserves_pending_input() {
        let active_turn = active.get_or_insert_with(ActiveTurn::default);
        Arc::clone(&active_turn.turn_state)
    };
-    turn_state
-        .lock()
-        .await
-        .push_pending_input(pending_item.clone());
+    sess.input_queue
+        .extend_pending_input_for_turn_state(turn_state.as_ref(), vec![pending_item.clone()])
+        .await;

    sess.abort_all_tasks(TurnAbortReason::Replaced).await;

    assert!(sess.active_turn.lock().await.is_none());
    assert_eq!(
-        turn_state.lock().await.take_pending_input(),
+        sess.input_queue
+            .take_pending_input_for_turn_state(turn_state.as_ref())
+            .await,
        vec![pending_item]
    );
 }
@@ -8593,7 +8601,7 @@ async fn budget_limited_accounting_steers_active_turn_without_aborting() -> anyh
    })
    .await?;

-    let pending_input = sess.get_pending_input().await;
+    let pending_input = sess.input_queue.get_pending_input(&sess.active_turn).await;
    let [ResponseInputItem::Message { role, content, .. }] = pending_input.as_slice() else {
        panic!("expected one budget-limit steering message, got {pending_input:#?}");
    };
@@ -8814,7 +8822,7 @@ async fn external_objective_change_steers_active_turn() -> anyhow::Result<()> {
    })
    .await?;

-    let pending_input = sess.get_pending_input().await;
+    let pending_input = sess.input_queue.get_pending_input(&sess.active_turn).await;
    assert!(
        pending_input.iter().any(|item| {
            matches!(
@@ -9021,19 +9029,26 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
    )
    .await;

-    sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await;
-    sess.enqueue_mailbox_communication(communication.clone());
+    sess.input_queue
+        .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
+        .await;
+    sess.input_queue
+        .enqueue_mailbox_communication(communication.clone())
+        .await;

    assert!(
-        !sess.has_pending_input().await,
+        !sess.input_queue.has_pending_input(&sess.active_turn).await,
        "queue-only mailbox mail should stay buffered once the current turn emitted its answer"
    );
-    assert_eq!(sess.get_pending_input().await, Vec::new());
+    assert_eq!(
+        sess.input_queue.get_pending_input(&sess.active_turn).await,
+        Vec::new()
+    );

    sess.abort_all_tasks(TurnAbortReason::Replaced).await;

    assert_eq!(
-        sess.get_pending_input().await,
+        sess.input_queue.get_pending_input(&sess.active_turn).await,
        vec![communication.to_response_input_item()],
    );
 }
@@ -9051,23 +9066,27 @@ async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
    )
    .await;

-    sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await;
-    sess.enqueue_mailbox_communication(InterAgentCommunication::new(
-        AgentPath::try_from("/root/worker").expect("worker path should parse"),
-        AgentPath::root(),
-        Vec::new(),
-        "late trigger update".to_string(),
-        /*trigger_turn*/ true,
-    ));
+    sess.input_queue
+        .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
+        .await;
+    sess.input_queue
+        .enqueue_mailbox_communication(InterAgentCommunication::new(
+            AgentPath::try_from("/root/worker").expect("worker path should parse"),
+            AgentPath::root(),
+            Vec::new(),
+            "late trigger update".to_string(),
+            /*trigger_turn*/ true,
+        ))
+        .await;

    assert!(
-        !sess.has_pending_input().await,
+        !sess.input_queue.has_pending_input(&sess.active_turn).await,
        "trigger-turn mailbox mail should not extend the current turn after its answer boundary"
    );

    sess.abort_all_tasks(TurnAbortReason::Replaced).await;

-    assert!(sess.has_trigger_turn_mailbox_items().await);
+    assert!(sess.input_queue.has_trigger_turn_mailbox_items().await);
 }

 #[tokio::test]
@@ -9090,8 +9109,12 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
    )
    .await;

-    sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await;
-    sess.enqueue_mailbox_communication(communication.clone());
+    sess.input_queue
+        .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
+        .await;
+    sess.input_queue
+        .enqueue_mailbox_communication(communication.clone())
+        .await;
    sess.steer_input(
        vec![UserInput::Text {
            text: "follow up".to_string(),
@@ -9104,7 +9127,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
    .expect("steered input should be accepted");

    assert_eq!(
-        sess.get_pending_input().await,
+        sess.input_queue.get_pending_input(&sess.active_turn).await,
        vec![
            ResponseInputItem::from(vec![UserInput::Text {
                text: "follow up".to_string(),
@@ -9135,8 +9158,12 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
    )
    .await;

-    sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await;
-    sess.enqueue_mailbox_communication(communication.clone());
+    sess.input_queue
+        .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
+        .await;
+    sess.input_queue
+        .enqueue_mailbox_communication(communication.clone())
+        .await;
    sess.steer_input(
        vec![UserInput::Text {
            text: "follow up".to_string(),
@@ -9148,10 +9175,12 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
    .await
    .expect("steered input should be accepted");

-    sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await;
+    sess.input_queue
+        .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
+        .await;

    assert_eq!(
-        sess.get_pending_input().await,
+        sess.input_queue.get_pending_input(&sess.active_turn).await,
        vec![
            ResponseInputItem::from(vec![UserInput::Text {
                text: "follow up".to_string(),
@@ -9182,8 +9211,12 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
    )
    .await;

-    sess.defer_mailbox_delivery_to_next_turn(&tc.sub_id).await;
-    sess.enqueue_mailbox_communication(communication.clone());
+    sess.input_queue
+        .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
+        .await;
+    sess.input_queue
+        .enqueue_mailbox_communication(communication.clone())
+        .await;

    let item = ResponseItem::FunctionCall {
        id: None,
@@ -9207,7 +9240,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
    assert!(output.needs_follow_up);
    assert!(output.tool_future.is_some());
    assert_eq!(
-        sess.get_pending_input().await,
+        sess.input_queue.get_pending_input(&sess.active_turn).await,
        vec![communication.to_response_input_item()],
    );
 }
@@ -148,7 +148,7 @@ pub(crate) async fn run_turn(
    prewarmed_client_session: Option<ModelClientSession>,
    cancellation_token: CancellationToken,
 ) -> Option<String> {
-    if input.is_empty() && !sess.has_pending_input().await {
+    if input.is_empty() && !sess.input_queue.has_pending_input(&sess.active_turn).await {
        return None;
    }

@@ -413,7 +413,7 @@ pub(crate) async fn run_turn(
        // submitted through the UI while the model was running. Though the UI
        // may support this, the model might not.
        let pending_input = if can_drain_pending_input {
-            sess.get_pending_input().await
+            sess.input_queue.get_pending_input(&sess.active_turn).await
        } else {
            Vec::new()
        };
@@ -434,7 +434,10 @@ pub(crate) async fn run_turn(
                    } => {
                        let remaining_pending_input = pending_input_iter.collect::<Vec<_>>();
                        if !remaining_pending_input.is_empty() {
-                            let _ = sess.prepend_pending_input(remaining_pending_input).await;
+                            let _ = sess
+                                .input_queue
+                                .prepend_pending_input(&sess.active_turn, remaining_pending_input)
+                                .await;
                            requeued_pending_input = true;
                        }
                        blocked_pending_input_contexts = additional_contexts;
@@ -494,7 +497,7 @@ pub(crate) async fn run_turn(
                    last_agent_message: sampling_request_last_agent_message,
                } = sampling_request_output;
                can_drain_pending_input = true;
-                let has_pending_input = sess.has_pending_input().await;
+                let has_pending_input = sess.input_queue.has_pending_input(&sess.active_turn).await;
                let needs_follow_up = model_needs_follow_up || has_pending_input;
                let total_usage_tokens = sess.get_total_token_usage().await;
                let token_limit_reached = total_usage_tokens >= auto_compact_limit;
@@ -2065,7 +2068,7 @@ async fn try_run_sampling_request(
                }
                needs_follow_up |= output_result.needs_follow_up;
                // todo: remove before stabilizing multi-agent v2
-                if preempt_for_mailbox_mail && sess.mailbox_rx.lock().await.has_pending() {
+                if preempt_for_mailbox_mail && sess.input_queue.has_pending_mailbox_items().await {
                    break Ok(SamplingRequestResult {
                        needs_follow_up: true,
                        last_agent_message,