[codex] Move pending input into input queue

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

@@ -499,13 +499,12 @@ impl AgentControl {
                             agent_nickname: None,
                             agent_role: None,
                         });
-                    match self
-                        .resume_single_agent_from_rollout(
-                            config.clone(),
-                            child_thread_id,
-                            child_session_source,
-                        )
-                        .await
+                    match Box::pin(self.resume_single_agent_from_rollout(
+                        config.clone(),
+                        child_thread_id,
+                        child_session_source,
+                    ))
+                    .await
                     {
                         Ok(_) => true,
                         Err(err) => {

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

@@ -486,7 +486,13 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
 
     timeout(Duration::from_secs(5), async {
         loop {
-            if thread.codex.session.has_pending_input().await {
+            if thread
+                .codex
+                .session
+                .input_queue
+                .has_pending_input(&thread.codex.session.active_turn)
+                .await
+            {
                 break;
             }
             sleep(Duration::from_millis(10)).await;

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

@@ -1,14 +1,11 @@
 pub(crate) mod agent_resolver;
 pub(crate) mod control;
-pub(crate) mod mailbox;
 mod registry;
 pub(crate) mod role;
 pub(crate) mod status;
 
 pub(crate) use codex_protocol::protocol::AgentStatus;
 pub(crate) use control::AgentControl;
-pub(crate) use mailbox::Mailbox;
-pub(crate) use mailbox::MailboxReceiver;
 pub(crate) use registry::exceeds_thread_spawn_depth_limit;
 pub(crate) use registry::next_thread_spawn_depth;
 pub(crate) use status::agent_status_from_event;

codex-rs/core/src/codex_thread.rs

@@ -383,6 +383,7 @@ impl CodexThread {
         {
             self.codex
                 .session
+                .input_queue
                 .queue_response_items_for_next_turn(items)
                 .await;
             self.codex.session.maybe_start_turn_for_pending_work().await;

codex-rs/core/src/goals.rs

@@ -1306,12 +1306,9 @@ impl Session {
                 .await;
             return;
         }
-        {
-            let mut turn_state = turn_state.lock().await;
-            for item in candidate.items {
-                turn_state.push_pending_input(item);
-            }
-        }
+        self.input_queue
+            .extend_pending_input_for_turn_state(turn_state.as_ref(), candidate.items)
+            .await;
 
         let turn_context = self
             .new_default_turn_with_sub_id(uuid::Uuid::new_v4().to_string())
@@ -1349,11 +1346,15 @@ impl Session {
             tracing::debug!("skipping active goal continuation because a turn is already active");
             return None;
         }
-        if self.has_queued_response_items_for_next_turn().await {
+        if self
+            .input_queue
+            .has_queued_response_items_for_next_turn()
+            .await
+        {
             tracing::debug!("skipping active goal continuation because queued input exists");
             return None;
         }
-        if self.has_trigger_turn_mailbox_items().await {
+        if self.input_queue.has_trigger_turn_mailbox_items().await {
             tracing::debug!(
                 "skipping active goal continuation because trigger-turn mailbox input is pending"
             );
@@ -1390,8 +1391,11 @@ impl Session {
             return None;
         }
         if self.active_turn.lock().await.is_some()
-            || self.has_queued_response_items_for_next_turn().await
-            || self.has_trigger_turn_mailbox_items().await
+            || self
+                .input_queue
+                .has_queued_response_items_for_next_turn()
+                .await
+            || self.input_queue.has_trigger_turn_mailbox_items().await
         {
             tracing::debug!("skipping active goal continuation because pending work appeared");
             return None;

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

@@ -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;
     }
 }