From 47c4d6d4b22295161d0ec947e8092ab8d5dc54d1 Mon Sep 17 00:00:00 2001 From: Edward Frazer Date: Wed, 20 May 2026 03:06:16 +0000 Subject: [PATCH] fix: keep queued follow-ups behind accepted direct turns --- codex-rs/app-server/README.md | 2 +- .../request_processors/thread_lifecycle.rs | 17 +++- .../thread_queue_processor.rs | 15 +++- .../src/request_processors/turn_processor.rs | 15 ++++ codex-rs/app-server/src/thread_state.rs | 34 +++++++- .../app-server/tests/suite/v2/thread_queue.rs | 79 +++++++++++++++++++ 6 files changed, 154 insertions(+), 8 deletions(-) diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 183bdcd797..340a8f9ad9 100644 --- a/codex-rs/app-server/README.md +++ b/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. diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 117bd1a290..d2b8aa78a3 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/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 { diff --git a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs index f5f7000b19..693834cf29 100644 --- a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs +++ b/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, diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 562592e3ec..06c68db4ab 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/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; diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 32dfcc325d..0ca707e1ee 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -25,6 +25,15 @@ use tracing::error; type PendingInterruptQueue = Vec; +#[derive(Default)] +pub(crate) enum PendingTurnStarts { + #[default] + None, + WaitingForLifecycle { + turn_ids: HashSet, + }, +} + pub(crate) struct PendingThreadResumeRequest { pub(crate) request_id: ConnectionRequestId, pub(crate) history_items: Vec, @@ -71,6 +80,7 @@ pub(crate) struct TurnSummary { pub(crate) struct ThreadState { pub(crate) pending_interrupts: PendingInterruptQueue, pub(crate) pending_rollbacks: Option, + pub(crate) pending_turn_starts: PendingTurnStarts, pub(crate) turn_summary: TurnSummary, pub(crate) last_terminal_turn_id: Option, pub(crate) cancel_tx: Option>, @@ -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 } } diff --git a/codex-rs/app-server/tests/suite/v2/thread_queue.rs b/codex-rs/app-server/tests/suite/v2/thread_queue.rs index 75df3b0c82..3de111f866 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_queue.rs +++ b/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,