From cfeccd51a0cf79d6957093d4a59fb63d81fb6069 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Fri, 5 Jun 2026 16:03:54 +0200 Subject: [PATCH 1/3] feat: add agent execution slot accounting --- codex-rs/core/src/agent/control.rs | 43 ++++++++ codex-rs/core/src/agent/registry.rs | 116 +++++++++++++++++----- codex-rs/core/src/agent/registry_tests.rs | 17 +++- codex-rs/core/src/tasks/lifecycle.rs | 8 ++ codex-rs/core/src/tasks/mod.rs | 56 ++++++++--- 5 files changed, 196 insertions(+), 44 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 6c6650a55933..60dbaa545651 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -7,6 +7,7 @@ use crate::agent::status::is_final; use crate::codex_thread::ThreadConfigSnapshot; use crate::config::Config; use crate::session::emit_subagent_session_started; +use crate::session::session::Session; use crate::session_prefix::format_subagent_context_line; use crate::session_prefix::format_subagent_notification_message; use crate::shell_snapshot::ShellSnapshot; @@ -267,6 +268,48 @@ impl AgentControl { Ok(thread.subscribe_status()) } + pub(crate) async fn reserve_execution_slot_for_pending_turn( + &self, + session: &Session, + ) -> CodexResult> { + if session.multi_agent_version() != Some(MultiAgentVersion::V2) { + return Ok(None); + } + let metadata = self + .state + .agent_metadata_for_thread(session.thread_id) + .ok_or_else(|| { + CodexErr::Fatal(format!( + "missing agent metadata for V2 thread {}", + session.thread_id + )) + })?; + if metadata.agent_path.as_ref().is_some_and(AgentPath::is_root) { + return Ok(None); + } + let config = session.get_config().await; + let max_threads = config + .effective_agent_max_threads(MultiAgentVersion::V2) + .unwrap_or_default(); + self.state + .reserve_execution_slot(session.thread_id, max_threads) + .map(Some) + } + + pub(crate) async fn release_execution_slot_if_idle(&self, session: &Session) { + let Some(metadata) = self.state.agent_metadata_for_thread(session.thread_id) else { + return; + }; + if metadata.agent_path.as_ref().is_none_or(AgentPath::is_root) { + return; + } + if session.active_turn.lock().await.is_none() + && !session.input_queue.has_trigger_turn_mailbox_items().await + { + self.state.release_execution_slot(session.thread_id); + } + } + pub(crate) async fn format_environment_context_subagents( &self, parent_thread_id: ThreadId, diff --git a/codex-rs/core/src/agent/registry.rs b/codex-rs/core/src/agent/registry.rs index 43aca201bfa2..37e457e35cae 100644 --- a/codex-rs/core/src/agent/registry.rs +++ b/codex-rs/core/src/agent/registry.rs @@ -13,12 +13,7 @@ use std::sync::Mutex; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; -/// This structure is used to add some limits on the multi-agent capabilities for Codex. In -/// the current implementation, it limits: -/// * Total number of sub-agents (i.e. threads) per user session -/// -/// This structure is shared by all agents in the same user session (because the `AgentControl` -/// is). +/// Tracks logical agents and active execution slots for one multi-agent session tree. #[derive(Default)] pub(crate) struct AgentRegistry { active_agents: Mutex, @@ -28,6 +23,7 @@ pub(crate) struct AgentRegistry { #[derive(Default)] struct ActiveAgents { agent_tree: HashMap, + active_thread_ids: HashSet, used_agent_nicknames: HashSet, nickname_reset_count: usize, } @@ -97,27 +93,61 @@ impl AgentRegistry { } pub(crate) fn release_spawned_thread(&self, thread_id: ThreadId) { - let removed_counted_agent = { - let mut active_agents = self - .active_agents - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let removed_key = active_agents - .agent_tree - .iter() - .find_map(|(key, metadata)| (metadata.agent_id == Some(thread_id)).then_some(key)) - .cloned(); - removed_key - .and_then(|key| active_agents.agent_tree.remove(key.as_str())) - .is_some_and(|metadata| { - !metadata.agent_path.as_ref().is_some_and(AgentPath::is_root) - }) - }; - if removed_counted_agent { + let mut active_agents = self + .active_agents + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let removed_key = active_agents + .agent_tree + .iter() + .find_map(|(key, metadata)| (metadata.agent_id == Some(thread_id)).then_some(key)) + .cloned(); + if let Some(removed_key) = removed_key { + active_agents.agent_tree.remove(removed_key.as_str()); + } + if active_agents.active_thread_ids.remove(&thread_id) { self.total_count.fetch_sub(1, Ordering::AcqRel); } } + pub(crate) fn release_execution_slot(&self, thread_id: ThreadId) { + let mut active_agents = self + .active_agents + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let removed = active_agents.active_thread_ids.remove(&thread_id); + if removed { + self.total_count.fetch_sub(1, Ordering::AcqRel); + } + } + + pub(crate) fn reserve_execution_slot( + self: &Arc, + thread_id: ThreadId, + max_threads: usize, + ) -> Result { + let mut active_agents = self + .active_agents + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if active_agents.active_thread_ids.contains(&thread_id) { + return Ok(ExecutionReservation { + state: Arc::clone(self), + thread_id, + release_on_drop: false, + }); + } + if !self.try_increment_spawned(max_threads) { + return Err(CodexErr::AgentLimitReached { max_threads }); + } + active_agents.active_thread_ids.insert(thread_id); + Ok(ExecutionReservation { + state: Arc::clone(self), + thread_id, + release_on_drop: true, + }) + } + pub(crate) fn register_root_thread(&self, thread_id: ThreadId) { let mut active_agents = self .active_agents @@ -194,7 +224,7 @@ impl AgentRegistry { } } - fn register_spawned_thread(&self, agent_metadata: AgentMetadata) { + pub(crate) fn register_agent(&self, mut agent_metadata: AgentMetadata) { let Some(thread_id) = agent_metadata.agent_id else { return; }; @@ -210,9 +240,25 @@ impl AgentRegistry { if let Some(agent_nickname) = agent_metadata.agent_nickname.clone() { active_agents.used_agent_nicknames.insert(agent_nickname); } + if let Some(existing) = active_agents.agent_tree.get(key.as_str()) + && agent_metadata.last_task_message.is_none() + { + agent_metadata.last_task_message = existing.last_task_message.clone(); + } active_agents.agent_tree.insert(key, agent_metadata); } + fn commit_spawned_thread(&self, agent_metadata: AgentMetadata) { + if let Some(thread_id) = agent_metadata.agent_id { + self.active_agents + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .active_thread_ids + .insert(thread_id); + } + self.register_agent(agent_metadata); + } + fn reserve_agent_nickname(&self, names: &[&str], preferred: Option<&str>) -> Option { let mut active_agents = self .active_agents @@ -337,7 +383,7 @@ impl SpawnReservation { pub(crate) fn commit(mut self, agent_metadata: AgentMetadata) { self.reserved_agent_nickname = None; self.reserved_agent_path = None; - self.state.register_spawned_thread(agent_metadata); + self.state.commit_spawned_thread(agent_metadata); self.active = false; } } @@ -353,6 +399,26 @@ impl Drop for SpawnReservation { } } +pub(crate) struct ExecutionReservation { + state: Arc, + thread_id: ThreadId, + release_on_drop: bool, +} + +impl ExecutionReservation { + pub(crate) fn commit(mut self) { + self.release_on_drop = false; + } +} + +impl Drop for ExecutionReservation { + fn drop(&mut self) { + if self.release_on_drop { + self.state.release_execution_slot(self.thread_id); + } + } +} + #[cfg(test)] #[path = "registry_tests.rs"] mod tests; diff --git a/codex-rs/core/src/agent/registry_tests.rs b/codex-rs/core/src/agent/registry_tests.rs index fc172fb336df..44349de5c184 100644 --- a/codex-rs/core/src/agent/registry_tests.rs +++ b/codex-rs/core/src/agent/registry_tests.rs @@ -81,7 +81,7 @@ fn reservation_drop_releases_slot() { } #[test] -fn commit_holds_slot_until_release() { +fn completion_releases_slot_without_removing_logical_agent() { let registry = Arc::new(AgentRegistry::default()); let reservation = registry.reserve_spawn_slot(Some(1)).expect("reserve slot"); let thread_id = ThreadId::new(); @@ -96,10 +96,21 @@ fn commit_holds_slot_until_release() { }; assert_eq!(max_threads, 1); - registry.release_spawned_thread(thread_id); + let followup = registry + .reserve_execution_slot(thread_id, 1) + .expect("running agent should retain its slot"); + followup.commit(); + assert!(registry.reserve_spawn_slot(Some(1)).is_err()); + registry.release_execution_slot(thread_id); + assert_eq!( + registry + .agent_metadata_for_thread(thread_id) + .and_then(|metadata| metadata.agent_id), + Some(thread_id) + ); let reservation = registry .reserve_spawn_slot(Some(1)) - .expect("slot released after thread removal"); + .expect("slot released after agent completion"); drop(reservation); } diff --git a/codex-rs/core/src/tasks/lifecycle.rs b/codex-rs/core/src/tasks/lifecycle.rs index 782cc8c7deda..59c46e617da6 100644 --- a/codex-rs/core/src/tasks/lifecycle.rs +++ b/codex-rs/core/src/tasks/lifecycle.rs @@ -1,5 +1,6 @@ use codex_extension_api::ExtensionData; use codex_protocol::protocol::CodexErrorInfo; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TurnAbortReason; @@ -53,6 +54,13 @@ impl Session { }) .await; } + + if self.multi_agent_version() == Some(MultiAgentVersion::V2) { + self.services + .agent_control + .release_execution_slot_if_idle(self) + .await; + } } pub(super) async fn emit_turn_abort_lifecycle( diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index ef2ba0f16d1a..a3b9936ca91c 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -460,27 +460,48 @@ impl Session { /// /// The turn is created only when there is mailbox mail marked with `trigger_turn`, and only /// if the session is currently idle. - pub(crate) async fn maybe_start_turn_for_pending_work_with_sub_id( + pub(crate) fn maybe_start_turn_for_pending_work_with_sub_id( self: &Arc, sub_id: String, - ) { - if !self.input_queue.has_trigger_turn_mailbox_items().await { - return; - } - - { - let mut active_turn = self.active_turn.lock().await; - if active_turn.is_some() { + ) -> BoxFuture<'_, ()> { + Box::pin(async move { + if !self.input_queue.has_trigger_turn_mailbox_items().await { return; } - *active_turn = Some(ActiveTurn::default()); - } - let turn_context = self.new_default_turn_with_sub_id(sub_id).await; - self.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) - .await; - self.start_task(turn_context, Vec::new(), RegularTask::new()) - .await; + let execution_reservation = match self + .services + .agent_control + .reserve_execution_slot_for_pending_turn(self) + .await + { + Ok(reservation) => reservation, + Err(err) => { + warn!( + thread_id = %self.thread_id, + "pending agent turn could not reserve an execution slot: {err}" + ); + return; + } + }; + + { + let mut active_turn = self.active_turn.lock().await; + if active_turn.is_some() { + return; + } + *active_turn = Some(ActiveTurn::default()); + } + if let Some(execution_reservation) = execution_reservation { + execution_reservation.commit(); + } + + let turn_context = self.new_default_turn_with_sub_id(sub_id).await; + self.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) + .await; + self.start_task(turn_context, Vec::new(), RegularTask::new()) + .await; + }) } pub async fn abort_all_tasks(self: &Arc, reason: TurnAbortReason) { @@ -519,6 +540,7 @@ impl Session { } if reason == TurnAbortReason::Interrupted && aborted_turn { self.maybe_start_turn_for_pending_work().await; + self.emit_thread_idle_lifecycle_if_idle().await; } } @@ -566,6 +588,7 @@ impl Session { if reason == TurnAbortReason::Interrupted { self.maybe_start_turn_for_pending_work().await; + self.emit_thread_idle_lifecycle_if_idle().await; } true @@ -797,6 +820,7 @@ impl Session { { warn!("failed to apply goal runtime maybe-continue event: {err}"); } + self.maybe_start_turn_for_pending_work().await; self.emit_thread_idle_lifecycle_if_idle().await; } From bdb868f9e24da74a2b344297fd2f51fd69db3218 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Fri, 5 Jun 2026 16:19:57 +0200 Subject: [PATCH 2/3] fix: wake queued agent turns after slot release --- codex-rs/core/src/agent/control.rs | 31 +++++++- codex-rs/core/src/agent/control/spawn.rs | 4 + codex-rs/core/src/agent/registry.rs | 3 +- codex-rs/core/src/session/handlers.rs | 26 +++++-- codex-rs/core/src/tasks/mod.rs | 96 ++++++++++++++++-------- 5 files changed, 119 insertions(+), 41 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 60dbaa545651..32d01f518d50 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -268,7 +268,7 @@ impl AgentControl { Ok(thread.subscribe_status()) } - pub(crate) async fn reserve_execution_slot_for_pending_turn( + pub(crate) async fn reserve_execution_slot_for_turn_start( &self, session: &Session, ) -> CodexResult> { @@ -305,8 +305,35 @@ impl AgentControl { } if session.active_turn.lock().await.is_none() && !session.input_queue.has_trigger_turn_mailbox_items().await + && self.state.release_execution_slot(session.thread_id) { - self.state.release_execution_slot(session.thread_id); + let Ok(state) = self.upgrade() else { + return; + }; + for agent in self.state.live_agents() { + let Some(thread_id) = agent.agent_id else { + continue; + }; + if thread_id == session.thread_id { + continue; + } + let Ok(thread) = state.get_thread(thread_id).await else { + continue; + }; + if thread + .codex + .session + .input_queue + .has_trigger_turn_mailbox_items() + .await + { + thread + .codex + .session + .maybe_start_turn_for_pending_work() + .await; + } + } } } diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 7cf5d00b4d40..a516908082a0 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -581,6 +581,10 @@ impl AgentControl { // Resumed threads are re-registered in-memory and need the same listener // attachment path as freshly spawned threads. state.notify_thread_created(resumed_thread.thread_id); + if multi_agent_version == MultiAgentVersion::V2 { + self.release_execution_slot_if_idle(resumed_thread.thread.codex.session.as_ref()) + .await; + } if multi_agent_version != MultiAgentVersion::V2 { let child_reference = agent_metadata .agent_path diff --git a/codex-rs/core/src/agent/registry.rs b/codex-rs/core/src/agent/registry.rs index 37e457e35cae..75324f9ee6da 100644 --- a/codex-rs/core/src/agent/registry.rs +++ b/codex-rs/core/src/agent/registry.rs @@ -110,7 +110,7 @@ impl AgentRegistry { } } - pub(crate) fn release_execution_slot(&self, thread_id: ThreadId) { + pub(crate) fn release_execution_slot(&self, thread_id: ThreadId) -> bool { let mut active_agents = self .active_agents .lock() @@ -119,6 +119,7 @@ impl AgentRegistry { if removed { self.total_count.fetch_sub(1, Ordering::AcqRel); } + removed } pub(crate) fn reserve_execution_slot( diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 7730a30ad77e..6a032f8166f0 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -22,6 +22,7 @@ use crate::realtime_conversation::prefix_realtime_v2_text; use crate::review_prompts::resolve_review_request; use crate::session::spawn_review_thread; use crate::tasks::CompactTask; +use crate::tasks::StartTaskOutcome; use crate::tasks::UserShellCommandMode; use crate::tasks::UserShellCommandTask; use crate::tasks::execute_user_shell_command; @@ -273,13 +274,24 @@ pub(super) async fn user_input_or_turn_inner( client_id: client_user_message_id, }); } - sess.spawn_task( - Arc::clone(¤t_context), - task_input, - crate::tasks::RegularTask::new(), - ) - .await; - Some(accepted_items) + match sess + .spawn_task( + Arc::clone(¤t_context), + task_input, + crate::tasks::RegularTask::new(), + ) + .await + { + StartTaskOutcome::Started => Some(accepted_items), + StartTaskOutcome::Rejected(err) => { + sess.send_event_raw(Event { + id: sub_id, + msg: EventMsg::Error(err.to_error_event(/*message_prefix*/ None)), + }) + .await; + None + } + } } Err(err) => { sess.send_event_raw(Event { diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index a3b9936ca91c..ac0687cd1acb 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -33,6 +33,7 @@ use crate::session::turn_context::TurnContext; use crate::state::ActiveTurn; use crate::state::RunningTask; use crate::state::TaskKind; +use crate::state::TurnState; use codex_analytics::TurnTokenUsageFact; use codex_login::AuthManager; use codex_models_manager::manager::SharedModelsManager; @@ -42,6 +43,7 @@ use codex_otel::TURN_MEMORY_METRIC; use codex_otel::TURN_NETWORK_PROXY_METRIC; use codex_otel::TURN_TOKEN_USAGE_METRIC; use codex_otel::TURN_TOOL_CALL_METRIC; +use codex_protocol::error::CodexErr; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::MultiAgentVersion; @@ -70,6 +72,12 @@ pub(crate) enum InterruptedTurnHistoryMarker { Developer, } +#[derive(Debug)] +pub(crate) enum StartTaskOutcome { + Started, + Rejected(CodexErr), +} + impl InterruptedTurnHistoryMarker { pub(crate) fn from_config_and_version( config: &Config, @@ -307,7 +315,7 @@ impl Session { turn_context: Arc, input: Vec, task: T, - ) { + ) -> StartTaskOutcome { self.abort_all_tasks(TurnAbortReason::Replaced).await; self.clear_connector_selection().await; self.start_task(turn_context, input, task).await; @@ -318,7 +326,37 @@ impl Session { turn_context: Arc, input: Vec, task: T, - ) { + ) -> StartTaskOutcome { + let turn_state = { + let mut active = self.active_turn.lock().await; + let turn = active.get_or_insert_with(ActiveTurn::default); + if turn.task.is_some() { + return StartTaskOutcome::Rejected(CodexErr::InvalidRequest( + "thread already has an active turn".to_string(), + )); + } + Arc::clone(&turn.turn_state) + }; + let execution_reservation = match self + .services + .agent_control + .reserve_execution_slot_for_turn_start(self) + .await + { + Ok(reservation) => reservation, + Err(err) => { + self.clear_reserved_active_turn(&turn_state).await; + warn!( + thread_id = %self.thread_id, + "agent turn could not reserve an execution slot: {err}" + ); + return StartTaskOutcome::Rejected(err); + } + }; + if let Some(execution_reservation) = execution_reservation { + execution_reservation.commit(); + } + let task: Arc = Arc::new(task); let task_kind = task.kind(); let span_name = task.span_name(); @@ -351,12 +389,6 @@ impl Session { warn!("failed to apply goal runtime turn-start event: {err}"); } let pending_items = self.input_queue.get_pending_input(&self.active_turn).await; - let turn_state = { - let mut active = self.active_turn.lock().await; - let turn = active.get_or_insert_with(ActiveTurn::default); - debug_assert!(turn.task.is_none()); - Arc::clone(&turn.turn_state) - }; turn_state.lock().await.token_usage_at_turn_start = token_usage_at_turn_start.clone(); self.input_queue .extend_pending_input_for_turn_state(turn_state.as_ref(), pending_items) @@ -366,7 +398,12 @@ impl Session { let turn_extension_data = Arc::clone(&turn_context.extension_data); let mut active = self.active_turn.lock().await; - let turn = active.get_or_insert_with(ActiveTurn::default); + let Some(turn) = active.as_mut() else { + return StartTaskOutcome::Rejected(CodexErr::Fatal( + "active turn reservation was lost before task start".to_string(), + )); + }; + debug_assert!(Arc::ptr_eq(&turn.turn_state, &turn_state)); debug_assert!(turn.task.is_none()); let done_clone = Arc::clone(&done); let session_ctx = Arc::new(SessionTaskContext::new( @@ -442,6 +479,7 @@ impl Session { _timer: timer, }; turn.task = Some(running_task); + StartTaskOutcome::Started } /// Starts a regular turn when the session is idle and pending work is waiting. @@ -469,36 +507,22 @@ impl Session { return; } - let execution_reservation = match self - .services - .agent_control - .reserve_execution_slot_for_pending_turn(self) - .await - { - Ok(reservation) => reservation, - Err(err) => { - warn!( - thread_id = %self.thread_id, - "pending agent turn could not reserve an execution slot: {err}" - ); - return; - } - }; - - { + let turn_state = { let mut active_turn = self.active_turn.lock().await; if active_turn.is_some() { return; } - *active_turn = Some(ActiveTurn::default()); - } - if let Some(execution_reservation) = execution_reservation { - execution_reservation.commit(); - } + let active_turn = active_turn.get_or_insert_with(ActiveTurn::default); + Arc::clone(&active_turn.turn_state) + }; let turn_context = self.new_default_turn_with_sub_id(sub_id).await; self.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) .await; + if !self.input_queue.has_trigger_turn_mailbox_items().await { + self.clear_reserved_active_turn(&turn_state).await; + return; + } self.start_task(turn_context, Vec::new(), RegularTask::new()) .await; }) @@ -829,6 +853,16 @@ impl Session { active.take() } + async fn clear_reserved_active_turn(&self, turn_state: &Arc>) { + let mut active_turn_guard = self.active_turn.lock().await; + if let Some(active_turn) = active_turn_guard.as_ref() + && active_turn.task.is_none() + && Arc::ptr_eq(&active_turn.turn_state, turn_state) + { + *active_turn_guard = None; + } + } + pub(crate) async fn close_unified_exec_processes(&self) { self.services .unified_exec_manager From 50e00c14cf5e87c12caeee0e941f7688249150ff Mon Sep 17 00:00:00 2001 From: jif-oai Date: Fri, 5 Jun 2026 16:36:00 +0200 Subject: [PATCH 3/3] fix: simplify task start result --- codex-rs/core/src/agent/registry.rs | 7 +- codex-rs/core/src/goals.rs | 3 +- codex-rs/core/src/session/handlers.rs | 21 +- codex-rs/core/src/session/inject.rs | 3 +- codex-rs/core/src/session/review.rs | 2 +- codex-rs/core/src/session/tests.rs | 481 ++++++++++++++------------ codex-rs/core/src/tasks/mod.rs | 30 +- 7 files changed, 284 insertions(+), 263 deletions(-) diff --git a/codex-rs/core/src/agent/registry.rs b/codex-rs/core/src/agent/registry.rs index 75324f9ee6da..ef64ab5ab582 100644 --- a/codex-rs/core/src/agent/registry.rs +++ b/codex-rs/core/src/agent/registry.rs @@ -225,7 +225,7 @@ impl AgentRegistry { } } - pub(crate) fn register_agent(&self, mut agent_metadata: AgentMetadata) { + fn register_agent(&self, agent_metadata: AgentMetadata) { let Some(thread_id) = agent_metadata.agent_id else { return; }; @@ -241,11 +241,6 @@ impl AgentRegistry { if let Some(agent_nickname) = agent_metadata.agent_nickname.clone() { active_agents.used_agent_nicknames.insert(agent_nickname); } - if let Some(existing) = active_agents.agent_tree.get(key.as_str()) - && agent_metadata.last_task_message.is_none() - { - agent_metadata.last_task_message = existing.last_task_message.clone(); - } active_agents.agent_tree.insert(key, agent_metadata); } diff --git a/codex-rs/core/src/goals.rs b/codex-rs/core/src/goals.rs index 6e0663655c92..b731c2aa1621 100644 --- a/codex-rs/core/src/goals.rs +++ b/codex-rs/core/src/goals.rs @@ -1326,7 +1326,8 @@ impl Session { .await; return; } - self.start_task(turn_context, Vec::new(), RegularTask::new()) + let _ = self + .start_task(turn_context, Vec::new(), RegularTask::new()) .await; } diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 6a032f8166f0..b9787b1c34df 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -22,7 +22,6 @@ use crate::realtime_conversation::prefix_realtime_v2_text; use crate::review_prompts::resolve_review_request; use crate::session::spawn_review_thread; use crate::tasks::CompactTask; -use crate::tasks::StartTaskOutcome; use crate::tasks::UserShellCommandMode; use crate::tasks::UserShellCommandTask; use crate::tasks::execute_user_shell_command; @@ -282,8 +281,8 @@ pub(super) async fn user_input_or_turn_inner( ) .await { - StartTaskOutcome::Started => Some(accepted_items), - StartTaskOutcome::Rejected(err) => { + Ok(()) => Some(accepted_items), + Err(err) => { sess.send_event_raw(Event { id: sub_id, msg: EventMsg::Error(err.to_error_event(/*message_prefix*/ None)), @@ -365,12 +364,13 @@ pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command } let turn_context = sess.new_default_turn_with_sub_id(sub_id).await; - sess.spawn_task( - Arc::clone(&turn_context), - Vec::new(), - UserShellCommandTask::new(command), - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&turn_context), + Vec::new(), + UserShellCommandTask::new(command), + ) + .await; } pub async fn resolve_elicitation( @@ -499,7 +499,8 @@ pub async fn reload_user_config(sess: &Arc) { pub async fn compact(sess: &Arc, sub_id: String) { let turn_context = sess.new_default_turn_with_sub_id(sub_id).await; - sess.spawn_task(Arc::clone(&turn_context), Vec::new(), CompactTask) + let _ = sess + .spawn_task(Arc::clone(&turn_context), Vec::new(), CompactTask) .await; } diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs index 3f73cd96d086..3c3d0f9aca65 100644 --- a/codex-rs/core/src/session/inject.rs +++ b/codex-rs/core/src/session/inject.rs @@ -124,7 +124,8 @@ impl Session { input.into_iter().map(TurnInput::ResponseItem).collect(), ) .await; - self.start_task(turn_context, Vec::new(), RegularTask::new()) + let _ = self + .start_task(turn_context, Vec::new(), RegularTask::new()) .await; Ok(()) } diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index ff4c77aab9f2..4154c1181936 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -174,7 +174,7 @@ pub(super) async fn spawn_review_thread( // TODO(ccunningham): Review turns currently rely on `spawn_task` for TurnComplete but do not // emit a parent TurnStarted. Consider giving review a full parent turn lifecycle // (TurnStarted + TurnComplete) for consistency with other standalone tasks. - sess.spawn_task(tc.clone(), input, ReviewTask::new()).await; + let _ = sess.spawn_task(tc.clone(), input, ReviewTask::new()).await; // Announce entering review mode so UIs can switch modes. let review_request = ReviewRequest { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index a2023ea378cc..12ca618b37fb 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -305,12 +305,13 @@ async fn regular_turn_emits_turn_started_with_trace_id_without_waiting_for_start ), ) .await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - crate::tasks::RegularTask::new(), - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + crate::tasks::RegularTask::new(), + ) + .await; let first = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()) .await @@ -386,12 +387,13 @@ async fn interrupting_regular_turn_waiting_on_startup_prewarm_emits_turn_aborted ), ) .await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - crate::tasks::RegularTask::new(), - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + crate::tasks::RegularTask::new(), + ) + .await; let first = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()) .await @@ -2111,15 +2113,16 @@ async fn turn_start_lifecycle_exposes_turn_metadata_and_token_baseline() { }; let sess = Arc::new(session); - sess.spawn_task( - Arc::new(turn_context), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::new(turn_context), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; let actual = records @@ -6309,20 +6312,21 @@ async fn spawn_task_turn_span_inherits_dispatch_trace_context() { let captured_trace = Arc::new(std::sync::Mutex::new(None)); async { - sess.spawn_task( - Arc::clone(&tc), - vec![TurnInput::UserInput { - content: vec![UserInput::Text { - text: "hello".to_string(), - text_elements: Vec::new(), + let _ = sess + .spawn_task( + Arc::clone(&tc), + vec![TurnInput::UserInput { + content: vec![UserInput::Text { + text: "hello".to_string(), + text_elements: Vec::new(), + }], + client_id: None, }], - client_id: None, - }], - TraceCaptureTask { - captured_trace: Arc::clone(&captured_trace), - }, - ) - .await; + TraceCaptureTask { + captured_trace: Arc::clone(&captured_trace), + }, + ) + .await; } .instrument(dispatch_span) .await; @@ -6498,7 +6502,7 @@ async fn submission_loop_channel_close_aborts_active_turn_before_thread_stop_lif session.services.extensions = Arc::new(builder.build()); let session = Arc::new(session); - session + let _ = session .spawn_task( Arc::new(turn_context), Vec::new(), @@ -7138,15 +7142,16 @@ async fn spawn_task_does_not_update_previous_turn_settings_for_non_run_turn_task client_id: None, }]; - sess.spawn_task( - Arc::clone(&tc), - input, - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; assert_eq!(sess.previous_turn_settings().await, None); @@ -8461,7 +8466,8 @@ async fn guardian_auto_review_interrupts_after_three_consecutive_denials() { }], client_id: None, }]; - sess.spawn_task(Arc::clone(&tc), input, GuardianDeniedApprovalTask) + let _ = sess + .spawn_task(Arc::clone(&tc), input, GuardianDeniedApprovalTask) .await; let mut observed = Vec::new(); @@ -8495,15 +8501,16 @@ async fn guardian_helper_review_interrupts_after_three_consecutive_denials() { }], client_id: None, }]; - sess.spawn_task( - Arc::clone(&tc), - input, - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; let session_for_review = Arc::clone(&sess); let turn_for_review = Arc::clone(&tc); @@ -8558,15 +8565,16 @@ async fn abort_regular_task_emits_marker_before_turn_aborted() { }], client_id: None, }]; - sess.spawn_task( - Arc::clone(&tc), - input, - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; @@ -8599,15 +8607,16 @@ async fn abort_gracefully_emits_marker_before_turn_aborted() { }], client_id: None, }]; - sess.spawn_task( - Arc::clone(&tc), - input, - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; @@ -8640,15 +8649,16 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() }], client_id: None, }]; - sess.spawn_task( - Arc::clone(&tc), - input, - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; while rx.try_recv().is_ok() {} @@ -8783,7 +8793,7 @@ async fn task_finish_emits_thread_idle_lifecycle_after_active_turn_clears() { session.services.extensions = Arc::new(builder.build()); let session = Arc::new(session); - session + let _ = session .spawn_task(Arc::new(turn_context), Vec::new(), CompletingTask) .await; @@ -8834,15 +8844,16 @@ async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() { #[tokio::test] async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; let item = user_message("synthetic idle input"); let err = sess @@ -8916,15 +8927,16 @@ async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() #[tokio::test] async fn try_start_turn_if_idle_rejects_active_review_turn_without_injecting() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Review, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Review, + listen_to_cancellation_token: true, + }, + ) + .await; let item = user_message("synthetic idle input"); let err = sess @@ -8974,15 +8986,16 @@ async fn steer_input_enforces_expected_turn_id() { }], client_id: None, }]; - sess.spawn_task( - Arc::clone(&tc), - input, - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; let steer_input = vec![UserInput::Text { text: "steer".to_string(), @@ -9025,15 +9038,16 @@ async fn steer_input_rejects_non_regular_turns() { client_id: None, }]; let turn_context = sess.new_default_turn_with_sub_id("turn".to_string()).await; - sess.spawn_task( - turn_context, - input, - NeverEndingTask { - kind: task_kind, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + turn_context, + input, + NeverEndingTask { + kind: task_kind, + listen_to_cancellation_token: true, + }, + ) + .await; let steer_input = vec![UserInput::Text { text: "steer".to_string(), @@ -9066,15 +9080,16 @@ async fn steer_input_returns_active_turn_id() { }], client_id: None, }]; - sess.spawn_task( - Arc::clone(&tc), - input, - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; let steer_input = vec![UserInput::Text { text: "steer".to_string(), @@ -9142,15 +9157,16 @@ async fn interrupt_accounts_active_goal_without_pausing() -> anyhow::Result<()> ) .await?; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; set_total_token_usage(&sess, post_goal_token_usage()).await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; @@ -9510,15 +9526,16 @@ async fn budget_limited_accounting_steers_active_turn_without_aborting() -> anyh token_usage: TokenUsage::default(), }) .await?; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; while rx.try_recv().is_ok() {} set_total_token_usage( @@ -9617,15 +9634,16 @@ async fn usage_limit_runtime_stops_active_goal_and_prevents_idle_continuation() token_usage: TokenUsage::default(), }) .await?; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; set_total_token_usage(&sess, post_goal_token_usage()).await; sess.goal_runtime_apply(GoalRuntimeEvent::UsageLimitReached { @@ -9662,15 +9680,16 @@ async fn external_goal_mutation_accounts_active_turn_before_status_change() -> a }, ) .await?; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; set_total_token_usage(&sess, post_goal_token_usage()).await; sess.goal_runtime_apply(GoalRuntimeEvent::ExternalMutationStarting) @@ -9724,15 +9743,16 @@ async fn external_goal_mutation_accounts_active_turn_before_status_change() -> a #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn external_objective_change_steers_active_turn() -> anyhow::Result<()> { let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; let state_db = goal_test_state_db(sess.as_ref()).await?; let old_goal = state_db @@ -9790,15 +9810,16 @@ async fn external_objective_change_steers_active_turn() -> anyhow::Result<()> { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn external_active_goal_set_marks_current_turn_for_accounting() -> anyhow::Result<()> { let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; set_total_token_usage(&sess, post_goal_token_usage()).await; let state_db = goal_test_state_db(sess.as_ref()).await?; @@ -9961,15 +9982,16 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { "late queue-only update".to_string(), /*trigger_turn*/ false, ); - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) @@ -10000,15 +10022,16 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { #[tokio::test] async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) @@ -10043,15 +10066,16 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { "queued child update".to_string(), /*trigger_turn*/ false, ); - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) @@ -10097,15 +10121,16 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { "queued child update".to_string(), /*trigger_turn*/ false, ); - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) @@ -10155,15 +10180,16 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { "queued child update".to_string(), /*trigger_turn*/ false, ); - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + let _ = sess + .spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; sess.input_queue .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) @@ -10211,7 +10237,8 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() { }], client_id: None, }]; - sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new()) + let _ = sess + .spawn_task(Arc::clone(&tc), input, ReviewTask::new()) .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index ac0687cd1acb..610ba63b55f4 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -44,6 +44,7 @@ use codex_otel::TURN_NETWORK_PROXY_METRIC; use codex_otel::TURN_TOKEN_USAGE_METRIC; use codex_otel::TURN_TOOL_CALL_METRIC; use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::MultiAgentVersion; @@ -72,12 +73,6 @@ pub(crate) enum InterruptedTurnHistoryMarker { Developer, } -#[derive(Debug)] -pub(crate) enum StartTaskOutcome { - Started, - Rejected(CodexErr), -} - impl InterruptedTurnHistoryMarker { pub(crate) fn from_config_and_version( config: &Config, @@ -315,10 +310,10 @@ impl Session { turn_context: Arc, input: Vec, task: T, - ) -> StartTaskOutcome { + ) -> CodexResult<()> { self.abort_all_tasks(TurnAbortReason::Replaced).await; self.clear_connector_selection().await; - self.start_task(turn_context, input, task).await; + self.start_task(turn_context, input, task).await } pub(crate) async fn start_task( @@ -326,12 +321,12 @@ impl Session { turn_context: Arc, input: Vec, task: T, - ) -> StartTaskOutcome { + ) -> CodexResult<()> { let turn_state = { let mut active = self.active_turn.lock().await; let turn = active.get_or_insert_with(ActiveTurn::default); if turn.task.is_some() { - return StartTaskOutcome::Rejected(CodexErr::InvalidRequest( + return Err(CodexErr::InvalidRequest( "thread already has an active turn".to_string(), )); } @@ -350,12 +345,9 @@ impl Session { thread_id = %self.thread_id, "agent turn could not reserve an execution slot: {err}" ); - return StartTaskOutcome::Rejected(err); + return Err(err); } }; - if let Some(execution_reservation) = execution_reservation { - execution_reservation.commit(); - } let task: Arc = Arc::new(task); let task_kind = task.kind(); @@ -399,7 +391,7 @@ impl Session { let turn_extension_data = Arc::clone(&turn_context.extension_data); let mut active = self.active_turn.lock().await; let Some(turn) = active.as_mut() else { - return StartTaskOutcome::Rejected(CodexErr::Fatal( + return Err(CodexErr::Fatal( "active turn reservation was lost before task start".to_string(), )); }; @@ -479,7 +471,10 @@ impl Session { _timer: timer, }; turn.task = Some(running_task); - StartTaskOutcome::Started + if let Some(execution_reservation) = execution_reservation { + execution_reservation.commit(); + } + Ok(()) } /// Starts a regular turn when the session is idle and pending work is waiting. @@ -523,7 +518,8 @@ impl Session { self.clear_reserved_active_turn(&turn_state).await; return; } - self.start_task(turn_context, Vec::new(), RegularTask::new()) + let _ = self + .start_task(turn_context, Vec::new(), RegularTask::new()) .await; }) }