fix: keep queued follow-ups behind accepted direct turns

codex-rs/app-server/README.md

@@ -147,7 +147,7 @@ Example with notification opt-out:
 - `thread/goal/clear` — clear the current persisted goal for a materialized thread; returns whether a goal was removed and emits `thread/goal/cleared` when state changes.
 - `thread/goal/updated` — notification emitted whenever a thread goal changes; includes the full current goal.
 - `thread/goal/cleared` — notification emitted whenever a thread goal is removed.
-- `thread/queue/add` — experimental; persist a future `turn/start` request for a loaded thread. The queue stores the original `TurnStartParams`, dispatches the oldest pending row once the thread is idle, and emits `thread/queue/changed`.
+- `thread/queue/add` — experimental; persist a future `turn/start` request for a loaded thread. The queue stores the original `TurnStartParams`, dispatches the oldest pending row once the thread is idle unless an older failed row blocks FIFO order, and emits `thread/queue/changed`.
 - `thread/queue/list` — experimental; page through the visible queued turns for a thread with cursor/limit pagination. Pending and failed rows are visible; the short-lived dispatch claim is internal.
 - `thread/queue/delete` — experimental; remove a visible queued turn by id.
 - `thread/queue/reorder` — experimental; replace the visible queue order by queued-turn id.

codex-rs/app-server/src/request_processors/thread_lifecycle.rs

@@ -296,10 +296,14 @@ pub(super) async fn ensure_listener_task_running(
                     // Track the event before emitting any typed translations
                     // so thread-local state such as raw event opt-in stays
                     // synchronized with the conversation.
-                    let raw_events_enabled = {
+                    let (raw_events_enabled, pending_turn_starts_cleared) = {
                         let mut thread_state = thread_state.lock().await;
-                        thread_state.track_current_turn_event(&event.id, &event.msg);
-                        thread_state.experimental_raw_events
+                        let pending_turn_starts_cleared =
+                            thread_state.track_current_turn_event(&event.id, &event.msg);
+                        (
+                            thread_state.experimental_raw_events,
+                            pending_turn_starts_cleared,
+                        )
                     };
                     let subscribed_connection_ids = thread_state_manager
                         .subscribed_connection_ids(conversation_id)
@@ -351,6 +355,13 @@ pub(super) async fn ensure_listener_task_running(
                             .drain_thread_queue_after_terminal_turn(conversation_id)
                             .await;
                     }
+                    if pending_turn_starts_cleared
+                        && matches!(&event.msg, EventMsg::Error(_))
+                    {
+                        thread_queue_processor
+                            .drain_thread_queue_after_terminal_turn(conversation_id)
+                            .await;
+                    }
                 }
                 unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => {
                     if !unloading_watchers_open {

codex-rs/app-server/src/request_processors/thread_queue_processor.rs

@@ -1,6 +1,7 @@
 use super::*;
 
 const THREAD_QUEUE_LIST_DEFAULT_LIMIT: usize = 25;
+const THREAD_QUEUE_LIST_MAX_LIMIT: usize = 100;
 
 #[derive(Clone)]
 pub(crate) struct ThreadQueueRequestProcessor {
@@ -92,7 +93,7 @@ impl ThreadQueueRequestProcessor {
         let limit = params
             .limit
             .unwrap_or(THREAD_QUEUE_LIST_DEFAULT_LIMIT as u32)
-            .max(1) as usize;
+            .clamp(1, THREAD_QUEUE_LIST_MAX_LIMIT as u32) as usize;
         let records = self
             .state_db()?
             .list_visible_thread_queued_turns_page(thread_id, start, limit.saturating_add(1))
@@ -225,8 +226,16 @@ impl ThreadQueueRequestProcessor {
             return;
         }
         let thread_state = self.thread_state_manager.thread_state(thread_id).await;
-        if thread_state.lock().await.active_turn_snapshot().is_some() {
-            return;
+        {
+            let thread_state = thread_state.lock().await;
+            if thread_state.active_turn_snapshot().is_some()
+                || !matches!(
+                    thread_state.pending_turn_starts,
+                    crate::thread_state::PendingTurnStarts::None
+                )
+            {
+                return;
+            }
         }
         let record = match state_db.claim_head_thread_queued_turn(thread_id).await {
             Ok(Some(record)) => record,

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -377,6 +377,21 @@ impl TurnRequestProcessor {
                 self.request_trace_context(&request_id).await,
             )
             .await?;
+        let thread_state = self.thread_state_manager.thread_state(thread_id).await;
+        {
+            let mut thread_state = thread_state.lock().await;
+            match &mut thread_state.pending_turn_starts {
+                crate::thread_state::PendingTurnStarts::None => {
+                    thread_state.pending_turn_starts =
+                        crate::thread_state::PendingTurnStarts::WaitingForLifecycle {
+                            turn_ids: HashSet::from([response.turn.id.clone()]),
+                        };
+                }
+                crate::thread_state::PendingTurnStarts::WaitingForLifecycle { turn_ids } => {
+                    turn_ids.insert(response.turn.id.clone());
+                }
+            }
+        }
         self.outgoing
             .record_request_turn_id(&request_id, &response.turn.id)
             .await;

codex-rs/app-server/src/thread_state.rs

@@ -25,6 +25,15 @@ use tracing::error;
 
 type PendingInterruptQueue = Vec<ConnectionRequestId>;
 
+#[derive(Default)]
+pub(crate) enum PendingTurnStarts {
+    #[default]
+    None,
+    WaitingForLifecycle {
+        turn_ids: HashSet<String>,
+    },
+}
+
 pub(crate) struct PendingThreadResumeRequest {
     pub(crate) request_id: ConnectionRequestId,
     pub(crate) history_items: Vec<RolloutItem>,
@@ -71,6 +80,7 @@ pub(crate) struct TurnSummary {
 pub(crate) struct ThreadState {
     pub(crate) pending_interrupts: PendingInterruptQueue,
     pub(crate) pending_rollbacks: Option<ConnectionRequestId>,
+    pub(crate) pending_turn_starts: PendingTurnStarts,
     pub(crate) turn_summary: TurnSummary,
     pub(crate) last_terminal_turn_id: Option<String>,
     pub(crate) cancel_tx: Option<oneshot::Sender<()>>,
@@ -112,6 +122,7 @@ impl ThreadState {
             let _ = cancel_tx.send(());
         }
         self.listener_command_tx = None;
+        self.pending_turn_starts = PendingTurnStarts::None;
         self.current_turn_history.reset();
         self.listener_thread = None;
         self.watch_registration = WatchRegistration::default();
@@ -131,17 +142,38 @@ impl ThreadState {
         self.current_turn_history.active_turn_snapshot()
     }
 
-    pub(crate) fn track_current_turn_event(&mut self, event_turn_id: &str, event: &EventMsg) {
+    pub(crate) fn track_current_turn_event(
+        &mut self,
+        event_turn_id: &str,
+        event: &EventMsg,
+    ) -> bool {
         if let EventMsg::TurnStarted(payload) = event {
             self.turn_summary.started_at = payload.started_at;
         }
         self.current_turn_history.handle_event(event);
+        let pending_turn_start_resolved = matches!(
+            event,
+            EventMsg::TurnStarted(_)
+                | EventMsg::TurnAborted(_)
+                | EventMsg::TurnComplete(_)
+                | EventMsg::Error(_)
+        );
+        let pending_turn_starts_cleared = match &mut self.pending_turn_starts {
+            PendingTurnStarts::None => false,
+            PendingTurnStarts::WaitingForLifecycle { turn_ids } => {
+                pending_turn_start_resolved && turn_ids.remove(event_turn_id) && turn_ids.is_empty()
+            }
+        };
+        if pending_turn_starts_cleared {
+            self.pending_turn_starts = PendingTurnStarts::None;
+        }
         if matches!(event, EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_))
             && !self.current_turn_history.has_active_turn()
         {
             self.last_terminal_turn_id = Some(event_turn_id.to_string());
             self.current_turn_history.reset();
         }
+        pending_turn_starts_cleared
     }
 }
 

codex-rs/app-server/tests/suite/v2/thread_queue.rs

@@ -371,6 +371,85 @@ async fn queued_turns_stay_serial_after_the_first_dispatch_starts() -> Result<()
     Ok(())
 }
 
+#[tokio::test]
+async fn queued_turns_wait_for_a_just_accepted_direct_turn_to_become_visible() -> Result<()> {
+    let responses = vec![
+        create_shell_command_sse_response(
+            vec![
+                "python3".to_string(),
+                "-c".to_string(),
+                "print(42)".to_string(),
+            ],
+            /*workdir*/ None,
+            Some(5000),
+            "direct-turn-blocker",
+        )?,
+        create_final_assistant_message_sse_response("direct turn done")?,
+        create_final_assistant_message_sse_response("queued follow-up done")?,
+    ];
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+    let codex_home = TempDir::new()?;
+    write_queue_test_config(codex_home.path(), &server.uri(), "untrusted")?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    initialize_experimental(&mut mcp).await?;
+    let thread = start_thread(&mut mcp).await?;
+
+    let direct_turn_request_id = mcp
+        .send_turn_start_request(text_turn(&thread.id, "direct turn first"))
+        .await?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(direct_turn_request_id)),
+    )
+    .await??;
+
+    let queued_turn_id = queue_turn(&mut mcp, &thread.id, "queued turn after direct").await?;
+    assert_eq!(
+        list_queue_ids(&mut mcp, &thread.id).await?,
+        vec![queued_turn_id]
+    );
+
+    let approval_request = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_request_message(),
+    )
+    .await??;
+    let ServerRequest::CommandExecutionRequestApproval { request_id, .. } = approval_request else {
+        panic!("expected direct turn approval request to keep the direct turn open");
+    };
+    mcp.send_response(
+        request_id,
+        serde_json::to_value(CommandExecutionRequestApprovalResponse {
+            decision: CommandExecutionApprovalDecision::Accept,
+        })?,
+    )
+    .await?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+    assert!(list_queue_ids(&mut mcp, &thread.id).await?.is_empty());
+
+    let requests = server
+        .received_requests()
+        .await
+        .expect("failed to fetch received requests");
+    assert_eq!(requests.len(), 3);
+    assert!(
+        String::from_utf8_lossy(&requests[2].body).contains("queued turn after direct"),
+        "queued follow-up should become its own later model request"
+    );
+
+    Ok(())
+}
+
 async fn initialize_experimental(mcp: &mut McpProcess) -> Result<()> {
     timeout(
         DEFAULT_READ_TIMEOUT,