From ef8672f22b4b8beec6ae9129cc124be098f3e96a Mon Sep 17 00:00:00 2001 From: efrazer-oai Date: Wed, 20 May 2026 15:29:32 -0700 Subject: [PATCH] feat: dispatch queued turns from app-server --- .../schema/json/ServerNotification.json | 6 + .../codex_app_server_protocol.schemas.json | 6 + .../codex_app_server_protocol.v2.schemas.json | 6 + .../v2/ThreadQueueChangedNotification.json | 6 + .../v2/ThreadQueueChangedNotification.ts | 2 +- .../src/protocol/v2/thread.rs | 1 + codex-rs/app-server/README.md | 9 +- codex-rs/app-server/src/message_processor.rs | 2 + .../request_processors/thread_lifecycle.rs | 27 + .../request_processors/thread_processor.rs | 9 +- .../thread_queue_processor.rs | 290 +++++++++- .../src/request_processors/turn_processor.rs | 84 ++- codex-rs/app-server/src/thread_state.rs | 125 ++++- .../app-server/tests/suite/v2/thread_queue.rs | 494 ++++++++++++++++-- codex-rs/state/src/runtime/queued_turns.rs | 95 ++++ 15 files changed, 1092 insertions(+), 70 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 0116cfc131..1b1a53b5d4 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -4329,6 +4329,12 @@ }, "ThreadQueueChangedNotification": { "properties": { + "dispatchingQueuedTurnId": { + "type": [ + "string", + "null" + ] + }, "queuedTurns": { "items": { "$ref": "#/definitions/QueuedTurn" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index f5d2e32483..a41df13ef8 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -17064,6 +17064,12 @@ "ThreadQueueChangedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { + "dispatchingQueuedTurnId": { + "type": [ + "string", + "null" + ] + }, "queuedTurns": { "items": { "$ref": "#/definitions/v2/QueuedTurn" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 8a994843a5..42d6009069 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -14888,6 +14888,12 @@ "ThreadQueueChangedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { + "dispatchingQueuedTurnId": { + "type": [ + "string", + "null" + ] + }, "queuedTurns": { "items": { "$ref": "#/definitions/QueuedTurn" diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json index a98dc853ec..702e71f9fc 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json @@ -485,6 +485,12 @@ } }, "properties": { + "dispatchingQueuedTurnId": { + "type": [ + "string", + "null" + ] + }, "queuedTurns": { "items": { "$ref": "#/definitions/QueuedTurn" diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts index d407a3abb3..d90f58d15e 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts @@ -3,4 +3,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { QueuedTurn } from "./QueuedTurn"; -export type ThreadQueueChangedNotification = { threadId: string, queuedTurns: Array, }; +export type ThreadQueueChangedNotification = { threadId: string, queuedTurns: Array, dispatchingQueuedTurnId: string | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 2725e7805f..5d2273139a 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -881,6 +881,7 @@ pub struct ThreadQueueReorderResponse { pub struct ThreadQueueChangedNotification { pub thread_id: String, pub queued_turns: Vec, + pub dispatching_queued_turn_id: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 93e84449c1..a0c548cfc8 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -149,11 +149,11 @@ Example with notification opt-out: - `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/settings/updated` — experimental notification emitted to subscribed clients when a loaded thread’s effective next-turn settings change; includes `threadId` and the full `threadSettings`. -- `thread/queue/add` — experimental; persist a future turn submission for a loaded thread. The queue stores the turn-scoped submission payload and emits `thread/queue/changed`. -- `thread/queue/list` — experimental; page through the visible queued turns for a thread with cursor/limit pagination. +- `thread/queue/add` — experimental; persist a future turn submission for a loaded thread. The queue stores the message and turn-scoped context, dispatches the oldest pending row with the thread's current settings 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. -- `thread/queue/changed` — experimental notification emitted after visible queue state changes and when a resumed thread replays its queue snapshot. +- `thread/queue/changed` — experimental notification emitted after visible queue state changes, including restart recovery that surfaces an interrupted dispatch as failed. - `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`). - `thread/archive` — move a thread’s rollout file into the archived directory and attempt to move any spawned descendant thread rollout files; returns `{}` on success and emits `thread/archived` for each archived thread. - `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server keeps the thread loaded and unloads it only after it has had no subscribers and no thread activity for 30 minutes, then emits `thread/closed`. @@ -607,6 +607,7 @@ Experimental clients can store a later submission with `thread/queue/add`. The q } } } { "method": "thread/queue/changed", "params": { "threadId": "thr_123", + "dispatchingQueuedTurnId": null, "queuedTurns": [{ "id": "queued_123", "submission": { @@ -617,7 +618,7 @@ Experimental clients can store a later submission with `thread/queue/add`. The q } } ``` -Delete or reorder visible queue rows with `thread/queue/delete` and `thread/queue/reorder`; list order is authoritative for rendering. +If dispatch cannot hand the row to a real turn, the row stays visible with `status.type: "failed"`. Delete or reorder visible queue rows with `thread/queue/delete` and `thread/queue/reorder`; list order is authoritative for rendering. ### Example: Archive a thread diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 430f1a90a1..0e44e85ee7 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -429,6 +429,7 @@ impl MessageProcessor { config_manager.clone(), Arc::clone(&pending_thread_unloads), thread_state_manager.clone(), + state_db.clone(), thread_watch_manager.clone(), Arc::clone(&thread_list_state_permit), Arc::clone(&skills_watcher), @@ -439,6 +440,7 @@ impl MessageProcessor { config_manager.clone(), state_db.clone(), thread_state_manager.clone(), + turn_processor.clone(), ); let thread_processor = ThreadRequestProcessor::new( auth_manager.clone(), 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 f72cfb5a62..416dfd0ad5 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -1,4 +1,5 @@ use super::*; +use crate::thread_state::TerminalTurnOutcome; pub(super) const THREAD_UNLOADING_DELAY: Duration = Duration::from_secs(30 * 60); @@ -13,6 +14,7 @@ pub(super) struct ListenerTaskContext { pub(super) fallback_model_provider: String, pub(super) codex_home: PathBuf, pub(super) skills_watcher: Arc, + pub(super) thread_queue_processor: ThreadQueueRequestProcessor, } struct UnloadingState { @@ -260,6 +262,7 @@ pub(super) async fn ensure_listener_task_running( thread_list_state_permit, fallback_model_provider, codex_home, + thread_queue_processor, .. } = listener_task_context; let outgoing_for_task = Arc::clone(&outgoing); @@ -284,6 +287,7 @@ pub(super) async fn ensure_listener_task_running( &thread_watch_manager, &outgoing_for_task, &pending_thread_unloads, + &thread_queue_processor, listener_command, ) .await; @@ -339,6 +343,23 @@ pub(super) async fn ensure_listener_task_running( fallback_model_provider.clone(), ) .await; + let terminal_outcome = match &event.msg { + EventMsg::TurnComplete(_) => Some(TerminalTurnOutcome::Completed), + EventMsg::TurnAborted(_) => Some(TerminalTurnOutcome::Aborted), + EventMsg::Error(error) if error.affects_turn_status() => { + Some(TerminalTurnOutcome::Error) + } + _ => None, + }; + if let Some(outcome) = terminal_outcome { + thread_queue_processor + .settle_dispatch_after_terminal_turn( + conversation_id, + event.id.as_str(), + outcome, + ) + .await; + } } unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => { if !unloading_watchers_open { @@ -454,6 +475,7 @@ pub(super) async fn handle_thread_listener_command( thread_watch_manager: &ThreadWatchManager, outgoing: &Arc, pending_thread_unloads: &Arc>>, + thread_queue_processor: &ThreadQueueRequestProcessor, listener_command: ThreadListenerCommand, ) { match listener_command { @@ -467,6 +489,7 @@ pub(super) async fn handle_thread_listener_command( thread_watch_manager, outgoing, pending_thread_unloads, + thread_queue_processor, *resume_request, ) .await; @@ -524,6 +547,7 @@ pub(super) async fn handle_pending_thread_resume_request( thread_watch_manager: &ThreadWatchManager, outgoing: &Arc, pending_thread_unloads: &Arc>>, + thread_queue_processor: &ThreadQueueRequestProcessor, pending: crate::thread_state::PendingThreadResumeRequest, ) { let active_turn = { @@ -697,6 +721,9 @@ pub(super) async fn handle_pending_thread_resume_request( { tracing::warn!("failed to continue active goal after running-thread resume: {err}"); } + thread_queue_processor + .emit_resume_queue_snapshot_and_drain(conversation_id) + .await; } pub(super) async fn send_thread_goal_snapshot_notification( diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 2d868aeb6d..eb429526a9 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -789,6 +789,7 @@ impl ThreadRequestProcessor { fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.to_path_buf(), skills_watcher: Arc::clone(&self.skills_watcher), + thread_queue_processor: self.thread_queue_processor.clone(), } } @@ -889,6 +890,7 @@ impl ThreadRequestProcessor { fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.to_path_buf(), skills_watcher: Arc::clone(&self.skills_watcher), + thread_queue_processor: self.thread_queue_processor.clone(), }; let request_trace = request_context.request_trace(); let config_manager = self.config_manager.clone(); @@ -2647,9 +2649,10 @@ impl ThreadRequestProcessor { config_snapshot.active_permission_profile, ); let token_usage_thread = include_turns.then(|| thread.clone()); + let response_rollout_items = response_history.get_rollout_items(); let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() { match build_thread_resume_initial_turns_page( - &response_history.get_rollout_items(), + &response_rollout_items, thread.status.clone(), /*has_live_running_thread*/ false, /*active_turn*/ None, @@ -2693,7 +2696,7 @@ impl ThreadRequestProcessor { // rebuilding history only to attribute a replayed usage update. if let Some(token_usage_thread) = token_usage_thread { let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items( - &response_history.get_rollout_items(), + &response_rollout_items, token_usage_thread.turns.as_slice(), ); // The client needs restored usage before it starts another turn. @@ -2713,7 +2716,7 @@ impl ThreadRequestProcessor { .emit_resume_goal_snapshot_and_continue(thread_id, codex_thread.as_ref()) .await; self.thread_queue_processor - .emit_resume_queue_snapshot(thread_id) + .recover_resume_queue_snapshot_and_drain(thread_id, &response_rollout_items) .await; } Err(err) => { 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 9692e3fb5f..fb3fd9bde0 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,7 +1,9 @@ use super::*; +use crate::thread_state::TerminalTurnOutcome; const THREAD_QUEUE_LIST_DEFAULT_LIMIT: usize = 25; const THREAD_QUEUE_LIST_MAX_LIMIT: usize = 100; +const THREAD_QUEUE_MAX_SUBMISSION_BYTES: usize = MAX_USER_INPUT_TEXT_CHARS; #[derive(Clone)] pub(crate) struct ThreadQueueRequestProcessor { @@ -10,6 +12,7 @@ pub(crate) struct ThreadQueueRequestProcessor { config_manager: ConfigManager, state_db: Option, thread_state_manager: ThreadStateManager, + turn_processor: TurnRequestProcessor, } impl ThreadQueueRequestProcessor { @@ -19,6 +22,7 @@ impl ThreadQueueRequestProcessor { config_manager: ConfigManager, state_db: Option, thread_state_manager: ThreadStateManager, + turn_processor: TurnRequestProcessor, ) -> Self { Self { thread_manager, @@ -26,6 +30,7 @@ impl ThreadQueueRequestProcessor { config_manager, state_db, thread_state_manager, + turn_processor, } } @@ -70,6 +75,11 @@ impl ThreadQueueRequestProcessor { let payload = serde_json::to_vec(&submission).map_err(|err| { internal_error(format!("failed to serialize queued turn payload: {err}")) })?; + if payload.len() > THREAD_QUEUE_MAX_SUBMISSION_BYTES { + return Err(invalid_params(format!( + "Queued turn submission exceeds the maximum length of {THREAD_QUEUE_MAX_SUBMISSION_BYTES} bytes." + ))); + } let record = self .state_db()? .thread_queue() @@ -86,6 +96,7 @@ impl ThreadQueueRequestProcessor { ) .await; self.emit_thread_queue_changed(thread_id).await; + self.drain_thread_queue_if_idle(thread_id).await; Ok(None) } @@ -139,6 +150,7 @@ impl ThreadQueueRequestProcessor { .await; if deleted { self.emit_thread_queue_changed(thread_id).await; + self.drain_thread_queue_if_idle(thread_id).await; } Ok(None) } @@ -168,22 +180,266 @@ impl ThreadQueueRequestProcessor { }, ) .await; - self.send_thread_queue_changed(thread_id, queued_turns) - .await; + self.emit_thread_queue_changed(thread_id).await; + self.drain_thread_queue_if_idle(thread_id).await; Ok(None) } - pub(crate) async fn emit_resume_queue_snapshot(&self, thread_id: ThreadId) { + pub(crate) async fn recover_resume_queue_snapshot_and_drain( + &self, + thread_id: ThreadId, + rollout_items: &[RolloutItem], + ) { + if self.require_enabled().await.is_err() { + return; + } + for turn in codex_app_server_protocol::build_turns_from_rollout_items(rollout_items) { + let outcome = match turn.status { + TurnStatus::Completed => Some(TerminalTurnOutcome::Completed), + TurnStatus::Interrupted => Some(TerminalTurnOutcome::Aborted), + TurnStatus::Failed => Some(TerminalTurnOutcome::Error), + TurnStatus::InProgress => None, + }; + if let Some(outcome) = outcome { + self.settle_dispatch_claim(thread_id, turn.id.as_str(), outcome) + .await; + } + } + let Some(state_db) = self.state_db.as_ref() else { + return; + }; + let failure = turn_error("queued turn dispatch was interrupted while app-server restarted"); + let failure_json = match serde_json::to_vec(&failure) { + Ok(failure_json) => failure_json, + Err(err) => { + tracing::warn!("failed to serialize queued turn recovery failure: {err}"); + return; + } + }; + match state_db + .thread_queue() + .recover_dispatching_thread_queued_turns(thread_id, failure_json.as_slice()) + .await + { + Ok(_) => {} + Err(err) => { + tracing::warn!("failed to recover queued turns for thread {thread_id}: {err}"); + return; + } + } + self.emit_thread_queue_changed(thread_id).await; + self.drain_thread_queue_if_idle(thread_id).await; + } + + pub(crate) async fn emit_resume_queue_snapshot_and_drain(&self, thread_id: ThreadId) { if self.require_enabled().await.is_err() { return; } self.emit_thread_queue_changed(thread_id).await; + self.drain_thread_queue_if_idle(thread_id).await; + } + + pub(crate) async fn settle_dispatch_after_terminal_turn( + &self, + thread_id: ThreadId, + turn_id: &str, + outcome: TerminalTurnOutcome, + ) { + self.settle_dispatch_claim(thread_id, turn_id, outcome) + .await; + self.drain_thread_queue_if_idle(thread_id).await; + } + + async fn settle_dispatch_claim( + &self, + thread_id: ThreadId, + turn_id: &str, + outcome: TerminalTurnOutcome, + ) { + let Some(state_db) = self.state_db.as_ref() else { + return; + }; + let result = match outcome { + TerminalTurnOutcome::Completed => { + state_db + .thread_queue() + .remove_dispatching_thread_queued_turn(thread_id, turn_id) + .await + } + TerminalTurnOutcome::Aborted | TerminalTurnOutcome::Error => { + let message = match outcome { + TerminalTurnOutcome::Aborted => "queued turn was aborted", + TerminalTurnOutcome::Error => "queued turn failed before it started", + TerminalTurnOutcome::Completed => unreachable!(), + }; + let failure_json = match serde_json::to_vec(&turn_error(message)) { + Ok(failure_json) => failure_json, + Err(err) => { + tracing::warn!("failed to serialize queued turn failure: {err}"); + return; + } + }; + state_db + .thread_queue() + .mark_dispatching_thread_queued_turn_failed( + thread_id, + turn_id, + failure_json.as_slice(), + ) + .await + } + }; + match result { + Ok(true) => self.emit_thread_queue_changed(thread_id).await, + Ok(false) => {} + Err(err) => { + tracing::warn!( + "failed to settle queued dispatch claim for thread {thread_id}: {err}" + ); + } + } + } + + async fn drain_thread_queue_if_idle(&self, thread_id: ThreadId) { + if self.require_enabled().await.is_err() { + return; + } + let Some(state_db) = self.state_db.as_ref() else { + return; + }; + loop { + let Ok(thread) = self.thread_manager.get_thread(thread_id).await else { + return; + }; + if matches!(thread.agent_status().await, AgentStatus::Running) { + return; + } + let thread_state = self.thread_state_manager.thread_state(thread_id).await; + { + 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 + .thread_queue() + .claim_head_thread_queued_turn(thread_id) + .await + { + Ok(Some(record)) => record, + Ok(None) => return, + Err(err) => { + tracing::warn!("failed to claim queued turn for thread {thread_id}: {err}"); + return; + } + }; + self.emit_thread_queue_changed(thread_id).await; + let submission = match serde_json::from_slice::( + record.turn_submission_jsonb.as_slice(), + ) { + Ok(submission) => submission, + Err(err) => { + self.fail_dispatch( + thread_id, + record.queued_turn_id.as_str(), + turn_error(format!("queued turn payload could not be read: {err}")), + ) + .await; + return; + } + }; + match self + .turn_processor + .queued_turn_start(thread_id, submission) + .await + { + Ok(response) => { + let turn_id = response.turn.id; + match state_db + .thread_queue() + .set_dispatching_thread_queued_turn_turn_id( + record.queued_turn_id.as_str(), + turn_id.as_str(), + ) + .await + { + Ok(true) => {} + Ok(false) => { + tracing::warn!( + "queued turn {} lost its dispatch claim before turn {turn_id} was recorded", + record.queued_turn_id + ); + return; + } + Err(err) => { + tracing::warn!( + "failed to record dispatch turn {turn_id} for queued turn {}: {err}", + record.queued_turn_id + ); + return; + } + } + let thread_state = self.thread_state_manager.thread_state(thread_id).await; + let terminal_outcome = { + let thread_state = thread_state.lock().await; + thread_state.terminal_outcome_for_turn(turn_id.as_str()) + }; + if let Some(outcome) = terminal_outcome { + self.settle_dispatch_claim(thread_id, turn_id.as_str(), outcome) + .await; + continue; + } + return; + } + Err(err) => { + self.fail_dispatch( + thread_id, + record.queued_turn_id.as_str(), + turn_error(format!( + "queued turn could not start: {message}", + message = err.message + )), + ) + .await; + return; + } + } + } + } + + async fn fail_dispatch(&self, thread_id: ThreadId, queued_turn_id: &str, error: TurnError) { + let Some(state_db) = self.state_db.as_ref() else { + return; + }; + let failure_json = match serde_json::to_vec(&error) { + Ok(failure_json) => failure_json, + Err(err) => { + tracing::warn!("failed to serialize queued turn failure: {err}"); + return; + } + }; + match state_db + .thread_queue() + .mark_thread_queued_turn_failed(queued_turn_id, failure_json.as_slice()) + .await + { + Ok(true) => self.emit_thread_queue_changed(thread_id).await, + Ok(false) => tracing::warn!( + "queued turn {queued_turn_id} could not be marked failed because its dispatch claim disappeared" + ), + Err(err) => tracing::warn!("failed to mark queued turn {queued_turn_id} failed: {err}"), + } } async fn emit_thread_queue_changed(&self, thread_id: ThreadId) { - match self.list_visible_queued_turns(thread_id).await { - Ok(queued_turns) => { - self.send_thread_queue_changed(thread_id, queued_turns) + match self.queue_snapshot(thread_id).await { + Ok((queued_turns, dispatching_queued_turn_id)) => { + self.send_thread_queue_changed(thread_id, queued_turns, dispatching_queued_turn_id) .await; } Err(err) => { @@ -192,7 +448,12 @@ impl ThreadQueueRequestProcessor { } } - async fn send_thread_queue_changed(&self, thread_id: ThreadId, queued_turns: Vec) { + async fn send_thread_queue_changed( + &self, + thread_id: ThreadId, + queued_turns: Vec, + dispatching_queued_turn_id: Option, + ) { let subscribed_connection_ids = self .thread_state_manager .subscribed_connection_ids(thread_id) @@ -203,6 +464,7 @@ impl ThreadQueueRequestProcessor { ServerNotification::ThreadQueueChanged(ThreadQueueChangedNotification { thread_id: thread_id.to_string(), queued_turns, + dispatching_queued_turn_id, }), ) .await; @@ -225,6 +487,20 @@ impl ThreadQueueRequestProcessor { .map(queued_turn_from_state) .collect() } + + async fn queue_snapshot( + &self, + thread_id: ThreadId, + ) -> Result<(Vec, Option), JSONRPCErrorError> { + let queued_turns = self.list_visible_queued_turns(thread_id).await?; + let dispatching_queued_turn_id = self + .state_db()? + .thread_queue() + .dispatching_thread_queued_turn_id(thread_id) + .await + .map_err(|err| internal_error(format!("failed to read queued dispatch: {err}")))?; + Ok((queued_turns, dispatching_queued_turn_id)) + } } fn parse_queue_thread_id(thread_id: &str) -> Result { 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 70ce7cc322..b18f7780ae 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -13,6 +13,7 @@ pub(crate) struct TurnRequestProcessor { config_manager: ConfigManager, pending_thread_unloads: Arc>>, thread_state_manager: ThreadStateManager, + state_db: Option, thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc, skills_watcher: Arc, @@ -83,6 +84,7 @@ impl TurnRequestProcessor { config_manager: ConfigManager, pending_thread_unloads: Arc>>, thread_state_manager: ThreadStateManager, + state_db: Option, thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc, skills_watcher: Arc, @@ -97,6 +99,7 @@ impl TurnRequestProcessor { config_manager, pending_thread_unloads, thread_state_manager, + state_db, thread_watch_manager, thread_list_state_permit, skills_watcher, @@ -120,6 +123,27 @@ impl TurnRequestProcessor { .map(|response| Some(response.into())) } + pub(crate) async fn queued_turn_start( + &self, + thread_id: ThreadId, + submission: TurnSubmission, + ) -> Result { + Self::validate_v2_input_limit(&submission.input)?; + let thread_id_string = thread_id.to_string(); + let (_, thread) = self.load_thread(&thread_id_string).await?; + let params = TurnStartParams { + thread_id: thread_id_string, + input: submission.input, + responsesapi_client_metadata: submission.responsesapi_client_metadata, + additional_context: submission.additional_context, + environments: submission.environments, + output_schema: submission.output_schema, + ..Default::default() + }; + self.start_turn_from_params(thread_id, thread, params, /*trace_context*/ None) + .await + } + pub(crate) async fn thread_inject_items( &self, params: ThreadInjectItemsParams, @@ -407,7 +431,48 @@ impl TurnRequestProcessor { .inspect_err(|error| { self.track_error_response(&request_id, error, /*error_type*/ None); })?; + let response = self + .start_turn_from_params( + thread_id, + thread, + params, + 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; + let turn_lifecycle_already_visible = thread_state + .active_turn_snapshot() + .is_some_and(|turn| turn.id == response.turn.id) + || thread_state.last_terminal_turn_id.as_deref() == Some(response.turn.id.as_str()); + if !turn_lifecycle_already_visible { + 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; + Ok(response) + } + async fn start_turn_from_params( + &self, + thread_id: ThreadId, + thread: Arc, + params: TurnStartParams, + trace_context: Option, + ) -> Result { let environment_selections = self.parse_environment_selections(params.environments)?; // Map v2 input items to core input items. @@ -452,15 +517,11 @@ impl TurnRequestProcessor { let turn_id = thread .submit_user_input_with_client_user_message_id( turn_op, - self.request_trace_context(&request_id).await, + trace_context, client_user_message_id, ) .await - .map_err(|err| { - let error = internal_error(format!("failed to start turn: {err}")); - self.track_error_response(&request_id, &error, /*error_type*/ None); - error - })?; + .map_err(|err| internal_error(format!("failed to start turn: {err}")))?; if turn_has_input { let config_snapshot = thread.config_snapshot().await; @@ -474,9 +535,6 @@ impl TurnRequestProcessor { ); } - self.outgoing - .record_request_turn_id(&request_id, &turn_id) - .await; let turn = Turn { id: turn_id, items: vec![], @@ -1279,6 +1337,14 @@ impl TurnRequestProcessor { fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.to_path_buf(), skills_watcher: Arc::clone(&self.skills_watcher), + thread_queue_processor: ThreadQueueRequestProcessor::new( + Arc::clone(&self.thread_manager), + Arc::clone(&self.outgoing), + self.config_manager.clone(), + self.state_db.clone(), + self.thread_state_manager.clone(), + self.clone(), + ), } } diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 2d932f0326..44ef484420 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -26,6 +26,28 @@ use tracing::error; type PendingInterruptQueue = Vec; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TerminalTurnOutcome { + Completed, + Aborted, + Error, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct LastTerminalTurn { + turn_id: String, + outcome: TerminalTurnOutcome, +} + +#[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, @@ -74,8 +96,10 @@ 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, + last_terminal_turn: Option, pub(crate) cancel_tx: Option>, pub(crate) experimental_raw_events: bool, pub(crate) listener_generation: u64, @@ -118,6 +142,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(); @@ -137,17 +162,54 @@ 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 terminal_outcome_for_turn(&self, turn_id: &str) -> Option { + self.last_terminal_turn + .as_ref() + .filter(|terminal_turn| terminal_turn.turn_id == turn_id) + .map(|terminal_turn| terminal_turn.outcome) + } + + 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); - if matches!(event, EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_)) - && !self.current_turn_history.has_active_turn() - { + let pending_turn_start_resolved = matches!( + event, + EventMsg::TurnStarted(_) | EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_) + ) || matches!(event, EventMsg::Error(error) if error.affects_turn_status()); + 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; + } + let terminal_outcome = match event { + EventMsg::TurnComplete(_) if !self.current_turn_history.has_active_turn() => { + Some(TerminalTurnOutcome::Completed) + } + EventMsg::TurnAborted(_) => Some(TerminalTurnOutcome::Aborted), + EventMsg::Error(error) if error.affects_turn_status() => { + Some(TerminalTurnOutcome::Error) + } + _ => None, + }; + if let Some(outcome) = terminal_outcome { + self.last_terminal_turn = Some(LastTerminalTurn { + turn_id: event_turn_id.to_string(), + outcome, + }); self.last_terminal_turn_id = Some(event_turn_id.to_string()); self.current_turn_history.reset(); } + pending_turn_starts_cleared } pub(crate) fn note_thread_settings(&mut self, thread_settings: ThreadSettings) -> bool { @@ -198,8 +260,63 @@ mod tests { use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; + use codex_protocol::protocol::ErrorEvent; + use codex_protocol::protocol::TurnAbortReason; + use codex_protocol::protocol::TurnAbortedEvent; + use codex_protocol::protocol::TurnStartedEvent; use pretty_assertions::assert_eq; + fn turn_started(turn_id: &str) -> EventMsg { + EventMsg::TurnStarted(TurnStartedEvent { + turn_id: turn_id.to_string(), + trace_id: None, + started_at: None, + model_context_window: None, + collaboration_mode_kind: Default::default(), + }) + } + + #[test] + fn aborted_turn_clears_active_snapshot() { + let mut state = ThreadState::default(); + state.track_current_turn_event("turn-1", &turn_started("turn-1")); + state.track_current_turn_event( + "turn-1", + &EventMsg::TurnAborted(TurnAbortedEvent { + turn_id: Some("turn-1".to_string()), + reason: TurnAbortReason::Interrupted, + completed_at: None, + duration_ms: None, + }), + ); + + assert!(state.active_turn_snapshot().is_none()); + assert_eq!( + state.terminal_outcome_for_turn("turn-1"), + Some(TerminalTurnOutcome::Aborted) + ); + } + + #[test] + fn status_affecting_error_clears_active_snapshot() { + let mut state = ThreadState::default(); + state.track_current_turn_event("turn-1", &turn_started("turn-1")); + state.track_current_turn_event( + "turn-1", + &EventMsg::Error(ErrorEvent { + message: "boom".to_string(), + codex_error_info: None, + }), + ); + + assert!(state.active_turn_snapshot().is_none()); + assert_eq!( + state.terminal_outcome_for_turn("turn-1"), + Some(TerminalTurnOutcome::Error) + ); + assert_eq!(state.last_terminal_turn_id.as_deref(), Some("turn-1")); + } + #[test] fn note_thread_settings_reports_only_effective_changes() { let mut state = ThreadState::default(); 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 463b8569b0..8eb3587105 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_queue.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_queue.rs @@ -1,9 +1,15 @@ use anyhow::Result; use app_test_support::McpProcess; use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_responses_server_sequence; use app_test_support::create_mock_responses_server_sequence_unchecked; +use app_test_support::create_shell_command_sse_response; use app_test_support::to_response; +use codex_app_server_protocol::AdditionalContextEntry; +use codex_app_server_protocol::AdditionalContextKind; use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::CommandExecutionApprovalDecision; +use codex_app_server_protocol::CommandExecutionRequestApprovalResponse; use codex_app_server_protocol::ExperimentalFeatureEnablementSetParams; use codex_app_server_protocol::ExperimentalFeatureEnablementSetResponse; use codex_app_server_protocol::InitializeCapabilities; @@ -11,6 +17,7 @@ use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::QueuedTurnStatus; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ThreadQueueAddParams; use codex_app_server_protocol::ThreadQueueAddResponse; use codex_app_server_protocol::ThreadQueueChangedNotification; @@ -22,9 +29,11 @@ use codex_app_server_protocol::ThreadQueueReorderParams; use codex_app_server_protocol::ThreadQueueReorderResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnSubmission; use codex_app_server_protocol::UserInput as V2UserInput; use std::collections::BTreeMap; +use std::collections::HashMap; use tempfile::TempDir; use tokio::time::timeout; @@ -32,13 +41,14 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const INVALID_REQUEST_ERROR_CODE: i64 = -32600; #[tokio::test] -async fn queue_add_persists_turn_params_and_emits_snapshot() -> Result<()> { - let server = create_mock_responses_server_sequence_unchecked(vec![ - create_final_assistant_message_sse_response("unused")?, - ]) - .await; +async fn idle_queue_add_dispatches_serialized_turn_and_drains_visible_queue() -> Result<()> { + let responses = vec![ + create_final_assistant_message_sse_response("queued done")?, + create_final_assistant_message_sse_response("second queued done")?, + ]; + let server = create_mock_responses_server_sequence(responses).await; let codex_home = TempDir::new()?; - write_queue_test_config(codex_home.path(), &server.uri())?; + write_queue_test_config(codex_home.path(), &server.uri(), "never")?; let mut mcp = McpProcess::new(codex_home.path()).await?; initialize_experimental(&mut mcp).await?; @@ -60,24 +70,84 @@ async fn queue_add_persists_turn_params_and_emits_snapshot() -> Result<()> { .await??; let ThreadQueueAddResponse { queued_turn } = to_response(add_response)?; assert!(matches!(queued_turn.status, QueuedTurnStatus::Pending)); - - let notification = timeout( + let add_notification = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_notification_message("thread/queue/changed"), ) .await??; - let notification: ThreadQueueChangedNotification = - serde_json::from_value(notification.params.expect("thread/queue/changed params"))?; - assert_eq!(notification.thread_id, thread.id); - assert_eq!(notification.queued_turns, vec![queued_turn.clone()]); + let add_notification: ThreadQueueChangedNotification = serde_json::from_value( + add_notification + .params + .expect("thread/queue/changed params"), + )?; + assert_eq!(add_notification.thread_id, thread.id); + assert_eq!(add_notification.queued_turns, vec![queued_turn.clone()]); + assert_eq!(add_notification.dispatching_queued_turn_id, None); - let ThreadQueueListResponse { data, next_cursor } = list_queue_page( - &mut mcp, &thread.id, /*cursor*/ None, /*limit*/ None, + let drain_notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/queue/changed"), ) - .await?; - assert_eq!(data, vec![queued_turn]); + .await??; + let drain_notification: ThreadQueueChangedNotification = serde_json::from_value( + drain_notification + .params + .expect("thread/queue/changed params"), + )?; + assert_eq!(drain_notification.thread_id, thread.id); + assert!(drain_notification.queued_turns.is_empty()); + assert_eq!( + drain_notification.dispatching_queued_turn_id, + Some(queued_turn.id) + ); + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let list_request_id = mcp + .send_raw_request( + "thread/queue/list", + Some(serde_json::to_value(ThreadQueueListParams { + thread_id: thread.id.clone(), + cursor: None, + limit: None, + })?), + ) + .await?; + let list_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(list_request_id)), + ) + .await??; + let ThreadQueueListResponse { data, next_cursor } = to_response(list_response)?; + assert!(data.is_empty()); assert_eq!(next_cursor, None); + queue_turn(&mut mcp, &thread.id, "second queued serialized input").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(), 2); + assert!( + String::from_utf8_lossy(&requests[0].body).contains("queued serialized input"), + "queued turn payload should reach the model request after state round-trip" + ); + assert!( + String::from_utf8_lossy(&requests[1].body).contains("second queued serialized input"), + "a later queued turn should still drain after a fast terminal dispatch" + ); + Ok(()) } @@ -88,7 +158,7 @@ async fn queue_add_rejects_ephemeral_threads() -> Result<()> { ]) .await; let codex_home = TempDir::new()?; - write_queue_test_config(codex_home.path(), &server.uri())?; + write_queue_test_config(codex_home.path(), &server.uri(), "never")?; let mut mcp = McpProcess::new(codex_home.path()).await?; initialize_experimental(&mut mcp).await?; @@ -133,6 +203,49 @@ async fn queue_add_rejects_ephemeral_threads() -> Result<()> { Ok(()) } +#[tokio::test] +async fn queue_add_rejects_oversized_serialized_submission() -> Result<()> { + let server = create_mock_responses_server_sequence_unchecked(vec![ + create_final_assistant_message_sse_response("unused")?, + ]) + .await; + let codex_home = TempDir::new()?; + write_queue_test_config(codex_home.path(), &server.uri(), "never")?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + initialize_experimental(&mut mcp).await?; + let thread = start_thread(&mut mcp).await?; + let mut submission = text_submission("small prompt"); + submission.additional_context = Some(HashMap::from([( + "oversized".to_string(), + AdditionalContextEntry { + value: "x".repeat((1 << 20) + 1), + kind: AdditionalContextKind::Application, + }, + )])); + + let add_request_id = mcp + .send_raw_request( + "thread/queue/add", + Some(serde_json::to_value(ThreadQueueAddParams { + thread_id: thread.id, + submission: submission.into(), + })?), + ) + .await?; + let add_error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(add_request_id)), + ) + .await??; + + assert_eq!( + add_error.error.message, + "Queued turn submission exceeds the maximum length of 1048576 bytes." + ); + Ok(()) +} + #[tokio::test] async fn queue_add_rejects_requests_when_feature_is_disabled() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(vec![ @@ -140,7 +253,7 @@ async fn queue_add_rejects_requests_when_feature_is_disabled() -> Result<()> { ]) .await; let codex_home = TempDir::new()?; - write_queue_test_config_without_feature(codex_home.path(), &server.uri())?; + write_queue_test_config_without_feature(codex_home.path(), &server.uri(), "never")?; let mut mcp = McpProcess::new(codex_home.path()).await?; initialize_experimental(&mut mcp).await?; @@ -172,17 +285,45 @@ async fn queue_add_rejects_requests_when_feature_is_disabled() -> Result<()> { #[tokio::test] async fn runtime_feature_enablement_controls_queue_access_without_deleting_rows() -> Result<()> { - let server = create_mock_responses_server_sequence_unchecked(vec![ - create_final_assistant_message_sse_response("unused")?, - ]) - .await; + let responses = vec![ + create_shell_command_sse_response( + vec![ + "python3".to_string(), + "-c".to_string(), + "print(42)".to_string(), + ], + /*workdir*/ None, + Some(5000), + "queue-feature-blocker", + )?, + create_final_assistant_message_sse_response("active turn done")?, + create_final_assistant_message_sse_response("queued turn done")?, + ]; + let server = create_mock_responses_server_sequence_unchecked(responses).await; let codex_home = TempDir::new()?; - write_queue_test_config_without_feature(codex_home.path(), &server.uri())?; + write_queue_test_config_without_feature(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 active_turn_request_id = mcp + .send_turn_start_request(text_turn(&thread.id, "keep the thread running")) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(active_turn_request_id)), + ) + .await??; + let approval_request = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_request_message(), + ) + .await??; + let ServerRequest::CommandExecutionRequestApproval { request_id, .. } = approval_request else { + panic!("expected command approval to keep the active turn open"); + }; + set_queue_feature(&mut mcp, /*enabled*/ true).await?; let queued_turn_id = queue_turn(&mut mcp, &thread.id, "durable queued turn").await?; @@ -214,29 +355,74 @@ async fn runtime_feature_enablement_controls_queue_access_without_deleting_rows( vec![queued_turn_id] ); + 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??; + Ok(()) } #[tokio::test] -async fn visible_queue_rows_support_pagination_reorder_and_delete() -> Result<()> { - let server = create_mock_responses_server_sequence_unchecked(vec![ - create_final_assistant_message_sse_response("unused")?, - ]) - .await; +async fn busy_thread_queue_rows_support_list_reorder_and_delete_before_drain() -> Result<()> { + let responses = vec![ + create_shell_command_sse_response( + vec![ + "python3".to_string(), + "-c".to_string(), + "print(42)".to_string(), + ], + /*workdir*/ None, + Some(5000), + "queue-blocker", + )?, + create_final_assistant_message_sse_response("active turn 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())?; + 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 active_turn_request_id = mcp + .send_turn_start_request(text_turn(&thread.id, "keep the thread running")) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(active_turn_request_id)), + ) + .await??; + + let approval_request = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_request_message(), + ) + .await??; + let ServerRequest::CommandExecutionRequestApproval { request_id, .. } = approval_request else { + panic!("expected command approval to keep the active turn open"); + }; + let first = queue_turn(&mut mcp, &thread.id, "first queued").await?; let second = queue_turn(&mut mcp, &thread.id, "second queued").await?; assert_eq!( list_queue_ids(&mut mcp, &thread.id).await?, vec![first.clone(), second.clone()] ); - let first_page = list_queue_page(&mut mcp, &thread.id, /*cursor*/ None, Some(1)).await?; assert_eq!( first_page @@ -285,6 +471,216 @@ async fn visible_queue_rows_support_pagination_reorder_and_delete() -> Result<() delete_queue_turn(&mut mcp, &thread.id, &first).await?; assert!(list_queue_ids(&mut mcp, &thread.id).await?.is_empty()); + 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??; + + Ok(()) +} + +#[tokio::test] +async fn queued_turns_stay_serial_after_the_first_dispatch_starts() -> Result<()> { + let responses = vec![ + create_shell_command_sse_response( + vec![ + "python3".to_string(), + "-c".to_string(), + "print(42)".to_string(), + ], + /*workdir*/ None, + Some(5000), + "queued-serial-blocker", + )?, + create_final_assistant_message_sse_response("first queued turn done")?, + create_final_assistant_message_sse_response("second queued turn 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?; + + queue_turn(&mut mcp, &thread.id, "first queued").await?; + let approval_request = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_request_message(), + ) + .await??; + let ServerRequest::CommandExecutionRequestApproval { request_id, .. } = approval_request else { + panic!("expected queued turn approval request to keep the first dispatch active"); + }; + + let second = queue_turn(&mut mcp, &thread.id, "second queued").await?; + assert_eq!(list_queue_ids(&mut mcp, &thread.id).await?, vec![second]); + + 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("second queued"), + "second queued follow-up should become its own later model request" + ); + + 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(()) +} + +#[tokio::test] +async fn queued_turns_drain_after_a_direct_turn_has_already_completed() -> Result<()> { + let responses = vec![ + 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(), "never")?; + + 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??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + queue_turn(&mut mcp, &thread.id, "queued turn after completion").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(), 2); + assert!( + String::from_utf8_lossy(&requests[1].body).contains("queued turn after completion"), + "queued follow-up should drain after an already completed direct turn" + ); + Ok(()) } @@ -426,30 +822,44 @@ fn text_submission(text: &str) -> TurnSubmission { } } -fn write_queue_test_config(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { - write_queue_test_config_with_feature(codex_home, server_uri, /*app_server_queue*/ true) -} - -fn write_queue_test_config_with_feature( - codex_home: &std::path::Path, - server_uri: &str, - app_server_queue: bool, -) -> std::io::Result<()> { - write_queue_test_config_with_optional_feature(codex_home, server_uri, Some(app_server_queue)) +fn text_turn(thread_id: &str, text: &str) -> TurnStartParams { + TurnStartParams { + thread_id: thread_id.to_string(), + input: text_submission(text).input, + ..Default::default() + } } fn write_queue_test_config_without_feature( codex_home: &std::path::Path, server_uri: &str, + approval_policy: &str, ) -> std::io::Result<()> { write_queue_test_config_with_optional_feature( - codex_home, server_uri, /*app_server_queue*/ None, + codex_home, + server_uri, + approval_policy, + /*app_server_queue*/ None, + ) +} + +fn write_queue_test_config( + codex_home: &std::path::Path, + server_uri: &str, + approval_policy: &str, +) -> std::io::Result<()> { + write_queue_test_config_with_optional_feature( + codex_home, + server_uri, + approval_policy, + /*app_server_queue*/ Some(true), ) } fn write_queue_test_config_with_optional_feature( codex_home: &std::path::Path, server_uri: &str, + approval_policy: &str, app_server_queue: Option, ) -> std::io::Result<()> { let feature_config = app_server_queue @@ -460,7 +870,7 @@ fn write_queue_test_config_with_optional_feature( format!( r#" model = "mock-model" -approval_policy = "never" +approval_policy = "{approval_policy}" sandbox_mode = "read-only" model_provider = "mock_provider" {feature_config} diff --git a/codex-rs/state/src/runtime/queued_turns.rs b/codex-rs/state/src/runtime/queued_turns.rs index e887d0af47..b5f7431f3f 100644 --- a/codex-rs/state/src/runtime/queued_turns.rs +++ b/codex-rs/state/src/runtime/queued_turns.rs @@ -261,6 +261,25 @@ RETURNING row.map(|row| thread_queued_turn_from_row(&row)).transpose() } + pub async fn dispatching_thread_queued_turn_id( + &self, + thread_id: ThreadId, + ) -> anyhow::Result> { + sqlx::query_scalar( + r#" +SELECT queued_turn_id +FROM thread_queued_turns +WHERE thread_id = ? + AND state = 'dispatching' +LIMIT 1 + "#, + ) + .bind(thread_id.to_string()) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(Into::into) + } + pub async fn set_dispatching_thread_queued_turn_turn_id( &self, queued_turn_id: &str, @@ -331,6 +350,35 @@ WHERE queued_turn_id = ? Ok(result.rows_affected() > 0) } + pub async fn mark_dispatching_thread_queued_turn_failed( + &self, + thread_id: ThreadId, + turn_id: &str, + failure_json: &[u8], + ) -> anyhow::Result { + let now_ms = datetime_to_epoch_millis(Utc::now()); + let result = sqlx::query( + r#" +UPDATE thread_queued_turns +SET + state = 'failed', + failure_jsonb = jsonb(?), + updated_at_ms = ? +WHERE thread_id = ? + AND state = 'dispatching' + AND dispatch_turn_id = ? + "#, + ) + .bind(failure_json) + .bind(now_ms) + .bind(thread_id.to_string()) + .bind(turn_id) + .execute(self.pool.as_ref()) + .await?; + + Ok(result.rows_affected() > 0) + } + pub async fn recover_dispatching_thread_queued_turns( &self, thread_id: ThreadId, @@ -427,6 +475,13 @@ mod tests { .expect("list visible queued turns"), Vec::new() ); + assert!( + queue + .dispatching_thread_queued_turn_id(thread_id) + .await + .expect("read dispatching queued turn id") + .is_some() + ); } #[tokio::test] @@ -655,6 +710,46 @@ mod tests { ); } + #[tokio::test] + async fn dispatch_claim_fails_only_for_its_submitted_turn() { + let (runtime, thread_id) = runtime_with_thread().await; + let queue = runtime.thread_queue(); + let queued_turn = queue + .append_thread_queued_turn(thread_id, br#"{"threadId":"t","input":[]}"#) + .await + .expect("append queued turn"); + queue + .claim_head_thread_queued_turn(thread_id) + .await + .expect("claim queued turn") + .expect("claimed row"); + queue + .set_dispatching_thread_queued_turn_turn_id(&queued_turn.queued_turn_id, "queued-turn") + .await + .expect("record submitted queued turn id"); + + assert!( + !queue + .mark_dispatching_thread_queued_turn_failed( + thread_id, + "regular-turn", + br#"{"message":"nope"}"#, + ) + .await + .expect("unmatched turn must not fail claim") + ); + assert!( + queue + .mark_dispatching_thread_queued_turn_failed( + thread_id, + "queued-turn", + br#"{"message":"nope"}"#, + ) + .await + .expect("matching queued turn fails claim") + ); + } + #[tokio::test] async fn deleting_thread_metadata_deletes_queued_turns() { let (runtime, thread_id) = runtime_with_thread().await;