From bf9fd885b2546e91fe3bab271aa481b070f47518 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 2 Jun 2026 14:31:00 +0200 Subject: [PATCH] Resolve per-thread multi-agent runtime (#25722) Stack split from #25708. Original PR intentionally left open. This third PR resolves the effective per-thread multi-agent runtime from persisted metadata, inherited runtime, and current model selection. --- codex-rs/core/src/agent/control.rs | 158 +++++++++++------- codex-rs/core/src/agent/control_tests.rs | 8 +- codex-rs/core/src/codex_delegate.rs | 2 + codex-rs/core/src/codex_thread.rs | 5 + codex-rs/core/src/config/config_tests.rs | 33 +++- codex-rs/core/src/config/mod.rs | 67 +++++--- codex-rs/core/src/session/mod.rs | 70 ++++++-- codex-rs/core/src/session/multi_agents.rs | 4 +- codex-rs/core/src/session/review.rs | 1 + codex-rs/core/src/session/session.rs | 9 +- codex-rs/core/src/session/tests.rs | 91 ++++++++++ .../core/src/session/tests/guardian_tests.rs | 2 +- codex-rs/core/src/session/turn_context.rs | 98 +++++++++-- codex-rs/core/src/session_startup_prewarm.rs | 2 +- codex-rs/core/src/tasks/mod.rs | 13 +- codex-rs/core/src/thread_manager.rs | 78 ++++++++- .../core/src/tools/handlers/agent_jobs.rs | 20 +-- .../handlers/multi_agents/resume_agent.rs | 2 +- .../src/tools/handlers/multi_agents/spawn.rs | 1 - .../src/tools/handlers/multi_agents_common.rs | 14 +- .../src/tools/handlers/multi_agents_tests.rs | 109 ++++++------ .../tools/handlers/multi_agents_v2/spawn.rs | 1 - codex-rs/core/src/tools/spec_plan.rs | 16 +- codex-rs/core/src/tools/spec_plan_tests.rs | 1 + 24 files changed, 599 insertions(+), 206 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index eecc46d58..821611e60 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -5,6 +5,7 @@ use crate::agent::role::DEFAULT_ROLE_NAME; use crate::agent::role::resolve_role_config; 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_prefix::format_subagent_context_line; use crate::session_prefix::format_subagent_notification_message; @@ -12,7 +13,6 @@ use crate::shell_snapshot::ShellSnapshot; use crate::thread_manager::ResumeThreadWithHistoryOptions; use crate::thread_manager::ThreadManagerState; use crate::thread_rollout_truncation::truncate_rollout_to_last_n_fork_turns; -use codex_features::Feature; use codex_protocol::AgentPath; use codex_protocol::SessionId; use codex_protocol::ThreadId; @@ -23,6 +23,7 @@ use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::RolloutItem; @@ -58,6 +59,11 @@ pub(crate) struct SpawnAgentOptions { pub(crate) environments: Option>, } +struct SpawnAgentThreadInheritance { + shell_snapshot: Option>, + exec_policy: Option>, +} + #[derive(Clone, Debug)] pub(crate) struct LiveAgent { pub(crate) thread_id: ThreadId, @@ -80,10 +86,7 @@ fn default_agent_nickname_list() -> Vec<&'static str> { .collect() } -fn agent_nickname_candidates( - config: &crate::config::Config, - role_name: Option<&str>, -) -> Vec { +fn agent_nickname_candidates(config: &Config, role_name: Option<&str>) -> Vec { let role_name = role_name.unwrap_or(DEFAULT_ROLE_NAME); if let Some(candidates) = resolve_role_config(config, role_name).and_then(|role| role.nickname_candidates.clone()) @@ -185,7 +188,7 @@ impl AgentControl { #[cfg(test)] pub(crate) async fn spawn_agent( &self, - config: crate::config::Config, + config: Config, initial_operation: Op, session_source: Option, ) -> CodexResult { @@ -202,7 +205,7 @@ impl AgentControl { /// Spawn an agent thread with some metadata. pub(crate) async fn spawn_agent_with_metadata( &self, - config: crate::config::Config, + config: Config, initial_operation: Op, session_source: Option, options: SpawnAgentOptions, // TODO(jif) drop with new fork. @@ -213,19 +216,33 @@ impl AgentControl { async fn spawn_agent_internal( &self, - config: crate::config::Config, + config: Config, initial_operation: Op, session_source: Option, options: SpawnAgentOptions, ) -> CodexResult { let state = self.upgrade()?; - let mut reservation = self.state.reserve_spawn_slot(config.agent_max_threads)?; - let inherited_shell_snapshot = self - .inherited_shell_snapshot_for_source(&state, session_source.as_ref()) - .await; - let inherited_exec_policy = self - .inherited_exec_policy_for_source(&state, session_source.as_ref(), &config) + let multi_agent_version = state + .effective_multi_agent_version_for_spawn( + &InitialHistory::New, + session_source.as_ref(), + options.parent_thread_id, + /*forked_from_thread_id*/ None, + &config, + ) .await; + let agent_max_threads = config + .effective_agent_max_threads(multi_agent_version) + .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?; + let mut reservation = self.state.reserve_spawn_slot(agent_max_threads)?; + let inheritance = SpawnAgentThreadInheritance { + shell_snapshot: self + .inherited_shell_snapshot_for_source(&state, session_source.as_ref()) + .await, + exec_policy: self + .inherited_exec_policy_for_source(&state, session_source.as_ref(), &config) + .await, + }; let (session_source, mut agent_metadata) = match session_source { Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, @@ -250,19 +267,19 @@ impl AgentControl { let notification_source = session_source.clone(); // The same `AgentControl` is sent to spawn the thread. - let new_thread = match (session_source, options.fork_mode.as_ref()) { - (Some(session_source), Some(_)) => { + let new_thread = match (session_source, options.fork_mode.as_ref(), inheritance) { + (Some(session_source), Some(_), inheritance) => { Box::pin(self.spawn_forked_thread( &state, config, session_source, &options, - inherited_shell_snapshot, - inherited_exec_policy, + inheritance, + multi_agent_version, )) .await? } - (Some(session_source), None) => { + (Some(session_source), None, inheritance) => { Box::pin(state.spawn_new_thread_with_source( config.clone(), self.clone(), @@ -271,13 +288,13 @@ impl AgentControl { /*forked_from_thread_id*/ None, /*thread_source*/ Some(ThreadSource::Subagent), /*metrics_service_name*/ None, - inherited_shell_snapshot, - inherited_exec_policy, + inheritance.shell_snapshot, + inheritance.exec_policy, options.environments.clone(), )) .await? } - (None, _) => Box::pin(state.spawn_new_thread(config.clone(), self.clone())).await?, + (None, _, _) => Box::pin(state.spawn_new_thread(config.clone(), self.clone())).await?, }; agent_metadata.agent_id = Some(new_thread.thread_id); reservation.commit(agent_metadata.clone()); @@ -340,7 +357,7 @@ impl AgentControl { self.send_input(new_thread.thread_id, initial_operation) .await?; - if !new_thread.thread.enabled(Feature::MultiAgentV2) { + if multi_agent_version != MultiAgentVersion::V2 { let child_reference = agent_metadata .agent_path .as_ref() @@ -364,12 +381,16 @@ impl AgentControl { async fn spawn_forked_thread( &self, state: &Arc, - config: crate::config::Config, + config: Config, session_source: SessionSource, options: &SpawnAgentOptions, - inherited_shell_snapshot: Option>, - inherited_exec_policy: Option>, + inheritance: SpawnAgentThreadInheritance, + multi_agent_version: MultiAgentVersion, ) -> CodexResult { + let SpawnAgentThreadInheritance { + shell_snapshot: inherited_shell_snapshot, + exec_policy: inherited_exec_policy, + } = inheritance; if options.fork_parent_spawn_call_id.is_none() { return Err(CodexErr::Fatal( "spawn_agent fork requires a parent spawn call id".to_string(), @@ -419,7 +440,7 @@ impl AgentControl { } let multi_agent_v2_usage_hint_texts_to_filter: Vec = if let Some(parent_thread) = parent_thread.as_ref() { - if parent_thread.enabled(Feature::MultiAgentV2) { + if multi_agent_version == MultiAgentVersion::V2 { let parent_config = parent_thread.codex.session.get_config().await; [ parent_config @@ -437,7 +458,7 @@ impl AgentControl { } else { Vec::new() } - } else if config.features.enabled(Feature::MultiAgentV2) { + } else if multi_agent_version == MultiAgentVersion::V2 { [ config.multi_agent_v2.root_agent_usage_hint_text.clone(), config.multi_agent_v2.subagent_usage_hint_text.clone(), @@ -473,7 +494,7 @@ impl AgentControl { } } if preserve_reference_context_item - && config.features.enabled(Feature::MultiAgentV2) + && multi_agent_version == MultiAgentVersion::V2 && config.multi_agent_v2.usage_hint_enabled && let Some(subagent_usage_hint_text) = config.multi_agent_v2.subagent_usage_hint_text.clone() @@ -504,7 +525,7 @@ impl AgentControl { /// Resume an existing agent thread from a recorded rollout file. pub(crate) async fn resume_agent_from_rollout( &self, - config: crate::config::Config, + config: Config, thread_id: ThreadId, session_source: SessionSource, ) -> CodexResult { @@ -579,20 +600,42 @@ impl AgentControl { async fn resume_single_agent_from_rollout( &self, - mut config: crate::config::Config, + config: Config, thread_id: ThreadId, session_source: SessionSource, ) -> CodexResult { - if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = &session_source - && *depth >= config.agent_max_depth - && !config.features.enabled(Feature::MultiAgentV2) - { - let _ = config.features.disable(Feature::SpawnCsv); - let _ = config.features.disable(Feature::Collab); - } let state = self.upgrade()?; let state_db_ctx = state.state_db(); - let mut reservation = self.state.reserve_spawn_slot(config.agent_max_threads)?; + let stored_thread = state + .read_stored_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: true, + }) + .await?; + let history = stored_thread + .history + .ok_or_else(|| CodexErr::ThreadNotFound(thread_id))? + .items; + let initial_history = InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history, + rollout_path: stored_thread.rollout_path, + }); + let parent_thread_id = stored_thread.parent_thread_id; + let multi_agent_version = state + .effective_multi_agent_version_for_spawn( + &initial_history, + Some(&session_source), + parent_thread_id, + /*forked_from_thread_id*/ None, + &config, + ) + .await; + let agent_max_threads = config + .effective_agent_max_threads(multi_agent_version) + .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?; + let mut reservation = self.state.reserve_spawn_slot(agent_max_threads)?; let (session_source, agent_metadata) = match session_source { SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, @@ -629,27 +672,11 @@ impl AgentControl { let inherited_exec_policy = self .inherited_exec_policy_for_source(&state, Some(&session_source), &config) .await; - let stored_thread = state - .read_stored_thread(ReadThreadParams { - thread_id, - include_archived: true, - include_history: true, - }) - .await?; - let history = stored_thread - .history - .ok_or_else(|| CodexErr::ThreadNotFound(thread_id))? - .items; - let parent_thread_id = stored_thread.parent_thread_id; let resumed_thread = state .resume_thread_with_history_with_source(ResumeThreadWithHistoryOptions { config: config.clone(), - initial_history: InitialHistory::Resumed(ResumedHistory { - conversation_id: thread_id, - history, - rollout_path: stored_thread.rollout_path, - }), + initial_history, agent_control: self.clone(), session_source, parent_thread_id, @@ -663,7 +690,7 @@ 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 !resumed_thread.thread.enabled(Feature::MultiAgentV2) { + if multi_agent_version != MultiAgentVersion::V2 { let child_reference = agent_metadata .agent_path .as_ref() @@ -1048,12 +1075,13 @@ impl AgentControl { }; let child_thread = state.get_thread(child_thread_id).await.ok(); let message = format_subagent_notification_message(child_reference.as_str(), &status); - if child_agent_path.is_some() - && child_thread - .as_ref() - .map(|thread| thread.enabled(Feature::MultiAgentV2)) - .unwrap_or(true) - { + let child_uses_multi_agent_v2 = match child_thread.as_ref() { + Some(child_thread) => { + child_thread.multi_agent_version() == Some(MultiAgentVersion::V2) + } + None => true, + }; + if child_agent_path.is_some() && child_uses_multi_agent_v2 { let Some(child_agent_path) = child_agent_path.clone() else { return; }; @@ -1089,7 +1117,7 @@ impl AgentControl { fn prepare_thread_spawn( &self, reservation: &mut crate::agent::registry::SpawnReservation, - config: &crate::config::Config, + config: &Config, parent_thread_id: ThreadId, depth: i32, agent_path: Option, @@ -1151,7 +1179,7 @@ impl AgentControl { &self, state: &Arc, session_source: Option<&SessionSource>, - child_config: &crate::config::Config, + child_config: &Config, ) -> Option> { let Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, .. diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 3e0242c03..f9462e94f 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -1587,9 +1587,15 @@ async fn spawn_child_completion_notifies_parent_history() { #[tokio::test] async fn multi_agent_v2_completion_ignores_dead_direct_parent() { let harness = AgentControlHarness::new().await; - let (root_thread_id, root_thread) = harness.start_thread().await; let mut config = harness.config.clone(); let _ = config.features.enable(Feature::MultiAgentV2); + let root = harness + .manager + .start_thread(config.clone()) + .await + .expect("root thread should start"); + let root_thread_id = root.thread_id; + let root_thread = root.thread; let worker_path = AgentPath::root().join("worker_a").expect("worker path"); let worker_thread_id = harness .control diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 74854b400..646d5969e 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -52,6 +52,7 @@ use codex_login::AuthManager; use codex_models_manager::manager::SharedModelsManager; use codex_protocol::error::CodexErr; use codex_protocol::protocol::InitialHistory; +use codex_protocol::protocol::MultiAgentVersion; #[cfg(test)] use crate::session::completed_session_loop_termination; @@ -103,6 +104,7 @@ pub(crate) async fn run_codex_thread_interactive( analytics_events_client: Some(parent_session.services.analytics_events_client.clone()), thread_store: Arc::clone(&parent_session.services.thread_store), attestation_provider: parent_session.services.attestation_provider.clone(), + inherited_multi_agent_version: Some(MultiAgentVersion::Disabled), })) .or_cancel(&cancel_token) .await??; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 7a6bc69c5..cd90829c0 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -24,6 +24,7 @@ use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AdditionalContextEntry; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::Event; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionConfiguredEvent; @@ -502,6 +503,10 @@ impl CodexThread { self.codex.session.get_config().await } + pub fn multi_agent_version(&self) -> Option { + self.codex.session.multi_agent_version() + } + /// Refresh the thread's layer-backed user config state from a caller-supplied /// config snapshot. Thread-scoped layers and session-static settings remain /// unchanged. diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 70719843b..05cbb9828 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -86,6 +86,7 @@ use codex_protocol::permissions::FileSystemSandboxEntry; use codex_protocol::permissions::FileSystemSandboxPolicy; use codex_protocol::permissions::FileSystemSpecialPath; use codex_protocol::permissions::NetworkSandboxPolicy; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::NetworkAccess; use codex_protocol::protocol::RealtimeVoice; use codex_protocol::protocol::SandboxPolicy; @@ -9801,7 +9802,13 @@ non_code_mode_only = true assert_eq!(config.multi_agent_v2.min_wait_timeout_ms, 2500); assert_eq!(config.multi_agent_v2.max_wait_timeout_ms, 120000); assert_eq!(config.multi_agent_v2.default_wait_timeout_ms, 30000); - assert_eq!(config.agent_max_threads, Some(4)); + assert_eq!( + ( + config.agent_max_threads, + config.effective_agent_max_threads(MultiAgentVersion::V2)? + ), + (None, Some(4)) + ); assert!(!config.multi_agent_v2.usage_hint_enabled); assert_eq!( config.multi_agent_v2.usage_hint_text.as_deref(), @@ -9845,7 +9852,13 @@ enabled = true assert_eq!(config.multi_agent_v2.min_wait_timeout_ms, 10_000); assert_eq!(config.multi_agent_v2.max_wait_timeout_ms, 3_600_000); assert_eq!(config.multi_agent_v2.default_wait_timeout_ms, 30_000); - assert_eq!(config.agent_max_threads, Some(3)); + assert_eq!( + ( + config.agent_max_threads, + config.effective_agent_max_threads(MultiAgentVersion::V2)? + ), + (None, Some(3)) + ); assert_eq!( config.multi_agent_v2.root_agent_usage_hint_text.as_deref(), Some(DEFAULT_MULTI_AGENT_V2_ROOT_AGENT_USAGE_HINT_TEXT) @@ -9912,17 +9925,19 @@ max_threads = 3 "#, )?; - let err = ConfigBuilder::without_managed_config_for_tests() + let config = ConfigBuilder::without_managed_config_for_tests() .codex_home(codex_home.path().to_path_buf()) .fallback_cwd(Some(codex_home.path().to_path_buf())) .build() - .await + .await?; + let err = config + .effective_agent_max_threads(MultiAgentVersion::V2) .expect_err("agents.max_threads should conflict with multi_agent_v2"); assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); assert_eq!( err.to_string(), - "agents.max_threads cannot be set when multi_agent_v2 is enabled" + "agents.max_threads cannot be set when the multi-agent runtime is v2" ); Ok(()) @@ -10181,7 +10196,13 @@ max_concurrent_threads_per_session = 1 .await?; assert_eq!(config.multi_agent_v2.max_concurrent_threads_per_session, 1); - assert_eq!(config.agent_max_threads, Some(0)); + assert_eq!( + ( + config.agent_max_threads, + config.effective_agent_max_threads(MultiAgentVersion::V2)? + ), + (None, Some(0)) + ); Ok(()) } diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index ff666b51e..051064f27 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -99,6 +99,7 @@ use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::permissions::FileSystemSandboxPolicy; use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SandboxPolicy; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::AbsolutePathBufGuard; @@ -823,7 +824,7 @@ pub struct Config { /// Token budget applied when storing tool/function outputs in the context manager. pub tool_output_token_limit: Option, - /// Maximum number of agent threads that can be open concurrently. + /// User-configured maximum number of agent threads that can be open concurrently. pub agent_max_threads: Option, /// Maximum runtime in seconds for agent job workers before they are failed. pub agent_job_max_runtime_seconds: Option, @@ -1281,6 +1282,40 @@ impl ConfigBuilder { } impl Config { + pub(crate) fn multi_agent_version_from_features(&self) -> MultiAgentVersion { + if self.features.enabled(Feature::MultiAgentV2) { + MultiAgentVersion::V2 + } else if self.features.enabled(Feature::Collab) { + MultiAgentVersion::V1 + } else { + MultiAgentVersion::Disabled + } + } + + pub(crate) fn effective_agent_max_threads( + &self, + multi_agent_version: MultiAgentVersion, + ) -> std::io::Result> { + match multi_agent_version { + MultiAgentVersion::V2 => { + if self.agent_max_threads.is_some() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "agents.max_threads cannot be set when the multi-agent runtime is v2", + )); + } + Ok(Some( + self.multi_agent_v2 + .max_concurrent_threads_per_session + .saturating_sub(1), + )) + } + MultiAgentVersion::Disabled | MultiAgentVersion::V1 => { + Ok(self.agent_max_threads.or(DEFAULT_AGENT_MAX_THREADS)) + } + } + } + pub fn legacy_sandbox_policy(&self) -> SandboxPolicy { self.permissions.legacy_sandbox_policy(self.cwd.as_path()) } @@ -3063,29 +3098,13 @@ impl Config { )); } validate_multi_agent_v2_tool_namespace(multi_agent_v2.tool_namespace.as_deref())?; - let agent_max_threads_from_config = cfg.agents.as_ref().and_then(|agents| agents.max_threads); - let agent_max_threads = if features.enabled(Feature::MultiAgentV2) { - if agent_max_threads_from_config.is_some() { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "agents.max_threads cannot be set when multi_agent_v2 is enabled", - )); - } - Some( - multi_agent_v2 - .max_concurrent_threads_per_session - .saturating_sub(1), - ) - } else { - let agent_max_threads = agent_max_threads_from_config.or(DEFAULT_AGENT_MAX_THREADS); - if agent_max_threads == Some(0) { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "agents.max_threads must be at least 1", - )); - } - agent_max_threads - }; + let agent_max_threads = cfg.agents.as_ref().and_then(|agents| agents.max_threads); + if agent_max_threads == Some(0) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "agents.max_threads must be at least 1", + )); + } let agent_max_depth = cfg .agents .as_ref() diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index e6e32a1b6..a63b0a91f 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -105,6 +105,7 @@ use codex_protocol::protocol::HasLegacyEvent; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::ItemStartedEvent; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::RawResponseItemEvent; use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::RolloutItem; @@ -418,6 +419,25 @@ pub(crate) struct CodexSpawnArgs { pub(crate) analytics_events_client: Option, pub(crate) thread_store: Arc, pub(crate) attestation_provider: Option>, + pub(crate) inherited_multi_agent_version: Option, +} + +pub(crate) fn resolve_multi_agent_version( + conversation_history: &InitialHistory, + inherited_multi_agent_version: Option, +) -> Option { + if inherited_multi_agent_version == Some(MultiAgentVersion::Disabled) { + return Some(MultiAgentVersion::Disabled); + } + + conversation_history + .get_multi_agent_version() + .or(inherited_multi_agent_version) + .or(match conversation_history { + InitialHistory::New | InitialHistory::Cleared => None, + // Threads created before runtime metadata existed keep the legacy V1 tool surface. + InitialHistory::Resumed(_) | InitialHistory::Forked(_) => Some(MultiAgentVersion::V1), + }) } pub(crate) const INITIAL_SUBMIT_ID: &str = ""; @@ -479,18 +499,11 @@ impl Codex { analytics_events_client, thread_store, attestation_provider, + inherited_multi_agent_version, } = args; let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_event, rx_event) = async_channel::unbounded(); - if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = session_source - && depth >= config.agent_max_depth - && !config.features.enabled(Feature::MultiAgentV2) - { - let _ = config.features.disable(Feature::SpawnCsv); - let _ = config.features.disable(Feature::Collab); - } - let primary_environment = environment_selections.primary_environment(); let mut user_instruction_warnings = Vec::new(); let user_instructions = AgentsMdManager::new(&config) @@ -541,6 +554,14 @@ impl Codex { let model_info = models_manager .get_model_info(model.as_str(), &config.to_models_manager_config()) .await; + let multi_agent_version = + resolve_multi_agent_version(&conversation_history, inherited_multi_agent_version); + let startup_multi_agent_version = multi_agent_version + .or(model_info.multi_agent_version) + .unwrap_or_else(|| config.multi_agent_version_from_features()); + let _ = config + .effective_agent_max_threads(startup_multi_agent_version) + .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?; let base_instructions = config .base_instructions .clone() @@ -625,6 +646,7 @@ impl Codex { thread_store, parent_rollout_thread_trace, attestation_provider, + multi_agent_version, ) .await .map_err(|e| { @@ -1166,7 +1188,6 @@ impl Session { } async fn record_initial_history(&self, conversation_history: InitialHistory) { - let turn_context = self.new_default_turn().await; let is_subagent = { let state = self.state.lock().await; state @@ -1187,6 +1208,7 @@ impl Session { .await; } InitialHistory::Resumed(resumed_history) => { + let turn_context = self.new_default_turn().await; let rollout_items = resumed_history.history; let previous_turn_settings = self .apply_rollout_reconstruction(&turn_context, &rollout_items) @@ -1226,6 +1248,7 @@ impl Session { } } InitialHistory::Forked(rollout_items) => { + let turn_context = self.new_default_turn().await; self.apply_rollout_reconstruction(&turn_context, &rollout_items) .await; @@ -1648,7 +1671,7 @@ impl Session { turn_context: &TurnContext, msg: &EventMsg, ) { - if !self.enabled(Feature::MultiAgentV2) { + if turn_context.multi_agent_version != MultiAgentVersion::V2 { return; } @@ -2619,6 +2642,33 @@ impl Session { state.session_configuration.collaboration_mode.clone() } + pub(crate) fn multi_agent_version(&self) -> Option { + self.multi_agent_version.get().copied() + } + + pub(crate) fn set_multi_agent_version_if_unset( + &self, + multi_agent_version: MultiAgentVersion, + ) -> MultiAgentVersion { + *self.multi_agent_version.get_or_init(|| multi_agent_version) + } + + pub(crate) fn resolve_multi_agent_version_for_model( + &self, + model_info: &ModelInfo, + config: &Config, + ) -> MultiAgentVersion { + if let Some(multi_agent_version) = self.multi_agent_version() { + return multi_agent_version; + } + + let selected = model_info + .multi_agent_version + .unwrap_or_else(|| config.multi_agent_version_from_features()); + + self.set_multi_agent_version_if_unset(selected) + } + async fn send_raw_response_items(&self, turn_context: &TurnContext, items: &[ResponseItem]) { for item in items { self.send_event( diff --git a/codex-rs/core/src/session/multi_agents.rs b/codex-rs/core/src/session/multi_agents.rs index eb3da0da4..a78a9c938 100644 --- a/codex-rs/core/src/session/multi_agents.rs +++ b/codex-rs/core/src/session/multi_agents.rs @@ -1,5 +1,5 @@ use crate::session::turn_context::TurnContext; -use codex_features::Feature; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; @@ -7,7 +7,7 @@ pub(super) fn usage_hint_text<'a>( turn_context: &'a TurnContext, session_source: &SessionSource, ) -> Option<&'a str> { - if !turn_context.features.enabled(Feature::MultiAgentV2) { + if turn_context.multi_agent_version != MultiAgentVersion::V2 { return None; } diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index c47643656..fe78e033e 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -128,6 +128,7 @@ pub(super) async fn spawn_review_thread( user_instructions: None, compact_prompt: parent_turn_context.compact_prompt.clone(), collaboration_mode: parent_turn_context.collaboration_mode.clone(), + multi_agent_version: MultiAgentVersion::Disabled, personality: parent_turn_context.personality, approval_policy: parent_turn_context.approval_policy.clone(), permission_profile: parent_turn_context.permission_profile(), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 966c96332..05773abb8 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -9,8 +9,10 @@ use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE; use codex_protocol::config_types::ServiceTier; use codex_protocol::permissions::FileSystemPath; use codex_protocol::permissions::FileSystemSpecialPath; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TurnEnvironmentSelection; +use std::sync::OnceLock; use tokio::sync::Semaphore; /// Context for an initialized model agent @@ -29,6 +31,7 @@ pub(crate) struct Session { /// The set of enabled features should be invariant for the lifetime of the /// session. pub(super) features: ManagedFeatures, + pub(super) multi_agent_version: OnceLock, pub(super) pending_mcp_server_refresh_config: Mutex>, pub(crate) conversation: Arc, pub(crate) active_turn: Mutex>, @@ -503,6 +506,7 @@ impl Session { thread_store: Arc, parent_rollout_thread_trace: ThreadTraceContext, attestation_provider: Option>, + multi_agent_version: Option, ) -> anyhow::Result> { debug!( "Configuring session: model={}; provider={:?}", @@ -517,6 +521,8 @@ impl Session { .parent_thread_id .or_else(|| initial_history.get_resumed_parent_thread_id()); session_configuration.parent_thread_id = parent_thread_id; + let multi_agent_version = multi_agent_version.map(OnceLock::from).unwrap_or_default(); + let initial_multi_agent_version = multi_agent_version.get().copied(); let thread_id = match &initial_history { InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => { @@ -558,7 +564,7 @@ impl Session { text: session_configuration.base_instructions.clone(), }, dynamic_tools: session_configuration.dynamic_tools.clone(), - multi_agent_version: None, + multi_agent_version: initial_multi_agent_version, metadata: ThreadPersistenceMetadata { cwd: Some(config.cwd.to_path_buf()), model_provider: config.model_provider_id.clone(), @@ -1063,6 +1069,7 @@ impl Session { state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), + multi_agent_version, pending_mcp_server_refresh_config: Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 28fa0ed27..d4b4942a8 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -112,6 +112,7 @@ use codex_protocol::protocol::CreditsSnapshot; use codex_protocol::protocol::GranularApprovalConfig; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::NetworkApprovalProtocol; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; @@ -121,6 +122,8 @@ use codex_protocol::protocol::RealtimeVoice; use codex_protocol::protocol::RealtimeVoicesList; use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::SessionMeta; +use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SkillScope; use codex_protocol::protocol::Submission; use codex_protocol::protocol::ThreadGoalStatus; @@ -176,6 +179,7 @@ use serde::Deserialize; use serde_json::json; use std::path::PathBuf; use std::sync::Arc; +use std::sync::OnceLock; use std::time::Duration as StdDuration; mod guardian_tests; @@ -1645,6 +1649,72 @@ async fn record_initial_history_reconstructs_resumed_transcript() { assert_eq!(expected, history.raw_items()); } +#[test] +fn resolve_multi_agent_version_handles_unset_and_legacy_history() { + let thread_id = ThreadId::default(); + + assert_eq!( + resolve_multi_agent_version( + &InitialHistory::New, + /*inherited_multi_agent_version*/ None + ), + None + ); + assert_eq!( + resolve_multi_agent_version( + &InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history: Vec::new(), + rollout_path: None, + }), + /*inherited_multi_agent_version*/ None, + ), + Some(MultiAgentVersion::V1) + ); + assert_eq!( + resolve_multi_agent_version( + &InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history: Vec::new(), + rollout_path: None, + }), + Some(MultiAgentVersion::V2), + ), + Some(MultiAgentVersion::V2) + ); + assert_eq!( + resolve_multi_agent_version( + &InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history: vec![session_meta_item( + thread_id, + Some(MultiAgentVersion::Disabled) + )], + rollout_path: None, + }), + Some(MultiAgentVersion::V2), + ), + Some(MultiAgentVersion::Disabled) + ); + assert_eq!( + resolve_multi_agent_version( + &InitialHistory::Forked(vec![session_meta_item( + thread_id, + Some(MultiAgentVersion::V2) + )]), + Some(MultiAgentVersion::Disabled), + ), + Some(MultiAgentVersion::Disabled) + ); + assert_eq!( + resolve_multi_agent_version( + &InitialHistory::Forked(Vec::new()), + /*inherited_multi_agent_version*/ None + ), + Some(MultiAgentVersion::V1) + ); +} + #[tokio::test] async fn record_initial_history_new_defers_initial_context_until_first_turn() { let (session, _turn_context) = make_session_and_context().await; @@ -1657,6 +1727,20 @@ async fn record_initial_history_new_defers_initial_context_until_first_turn() { assert_eq!(session.previous_turn_settings().await, None); } +fn session_meta_item( + thread_id: ThreadId, + multi_agent_version: Option, +) -> RolloutItem { + RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + id: thread_id, + multi_agent_version, + ..SessionMeta::default() + }, + git: None, + }) +} + #[tokio::test] async fn resumed_history_injects_initial_context_on_first_context_update_only() { let (session, turn_context) = make_session_and_context().await; @@ -4520,6 +4604,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, + Some(config.multi_agent_version_from_features()), ) .await; @@ -4718,6 +4803,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { &session_telemetry, session_configuration.provider.clone(), &session_configuration, + config.multi_agent_version_from_features(), services.user_shell.as_ref(), services.shell_zsh_path.as_ref(), services.main_execve_wrapper_exe.as_ref(), @@ -4741,6 +4827,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), + multi_agent_version: OnceLock::from(config.multi_agent_version_from_features()), pending_mcp_server_refresh_config: Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), @@ -4867,6 +4954,7 @@ async fn make_session_with_config_and_rx( )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, + Some(config.multi_agent_version_from_features()), ) .await?; @@ -4978,6 +5066,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, + Some(config.multi_agent_version_from_features()), ) .await?; @@ -6563,6 +6652,7 @@ where &session_telemetry, session_configuration.provider.clone(), &session_configuration, + config.multi_agent_version_from_features(), services.user_shell.as_ref(), services.shell_zsh_path.as_ref(), services.main_execve_wrapper_exe.as_ref(), @@ -6586,6 +6676,7 @@ where state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), + multi_agent_version: OnceLock::from(config.multi_agent_version_from_features()), pending_mcp_server_refresh_config: Mutex::new(None), conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index cd2548ba0..2977e6bdf 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -709,6 +709,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { analytics_events_client: None, thread_store, attestation_provider: None, + inherited_multi_agent_version: None, }) .await .expect("spawn guardian subagent"); @@ -728,6 +729,5 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { }], } ); - drop(codex); } diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index b9ed483f1..959398430 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -7,7 +7,9 @@ use codex_model_provider::create_model_provider; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::models::AdditionalPermissionProfile; +use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ToolMode; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_sandboxing::compatibility_sandbox_policy_for_permission_profile; @@ -78,6 +80,7 @@ pub struct TurnContext { pub(crate) compact_prompt: Option, pub(crate) user_instructions: Option, pub(crate) collaboration_mode: CollaborationMode, + pub(crate) multi_agent_version: MultiAgentVersion, pub(crate) personality: Option, pub(crate) approval_policy: Constrained, pub(crate) permission_profile: PermissionProfile, @@ -101,6 +104,12 @@ pub struct TurnContext { pub(crate) server_model_warning_emitted: AtomicBool, pub(crate) model_verification_emitted: AtomicBool, } + +enum TurnMultiAgentRuntime { + ResolveAndStore, + Preview, +} + impl TurnContext { pub(crate) fn permission_profile(&self) -> PermissionProfile { self.permission_profile.clone() @@ -246,6 +255,7 @@ impl TurnContext { compact_prompt: self.compact_prompt.clone(), user_instructions: self.user_instructions.clone(), collaboration_mode, + multi_agent_version: self.multi_agent_version, personality: self.personality, approval_policy: self.approval_policy.clone(), permission_profile: self.permission_profile.clone(), @@ -353,7 +363,7 @@ impl TurnContext { model: self.model_info.slug.clone(), personality: self.personality, collaboration_mode: Some(self.collaboration_mode.clone()), - multi_agent_version: None, + multi_agent_version: Some(self.multi_agent_version), realtime_active: Some(self.realtime_active), effort: self.reasoning_effort, summary: ReasoningSummaryConfig::Auto, @@ -455,6 +465,7 @@ impl Session { session_telemetry: &SessionTelemetry, provider: ModelProviderInfo, session_configuration: &SessionConfiguration, + multi_agent_version: MultiAgentVersion, user_shell: &shell::Shell, shell_zsh_path: Option<&PathBuf>, main_execve_wrapper_exe: Option<&PathBuf>, @@ -544,6 +555,7 @@ impl Session { compact_prompt: session_configuration.compact_prompt.clone(), user_instructions: session_configuration.user_instructions.clone(), collaboration_mode: session_configuration.collaboration_mode.clone(), + multi_agent_version, personality: session_configuration.personality, approval_policy: session_configuration.approval_policy.clone(), permission_profile: session_configuration.permission_profile(), @@ -685,8 +697,43 @@ impl Session { final_output_json_schema: Option>, turn_environments: ResolvedTurnEnvironments, ) -> Arc { - let primary_turn_environment = turn_environments.primary(); + self.new_turn_context_from_configuration( + sub_id, + session_configuration, + final_output_json_schema, + turn_environments, + TurnMultiAgentRuntime::ResolveAndStore, + ) + .await + } + + async fn new_startup_prewarm_turn_from_configuration( + &self, + sub_id: String, + session_configuration: SessionConfiguration, + turn_environments: ResolvedTurnEnvironments, + ) -> Arc { + self.new_turn_context_from_configuration( + sub_id, + session_configuration, + /*final_output_json_schema*/ None, + turn_environments, + TurnMultiAgentRuntime::Preview, + ) + .await + } + + async fn new_turn_context_from_configuration( + &self, + sub_id: String, + session_configuration: SessionConfiguration, + final_output_json_schema: Option>, + turn_environments: ResolvedTurnEnvironments, + multi_agent_runtime: TurnMultiAgentRuntime, + ) -> Arc { + let primary_turn_environment = turn_environments.primary().cloned(); let cwd = primary_turn_environment + .as_ref() .map(|turn_environment| turn_environment.cwd.clone()) .unwrap_or_else(|| session_configuration.cwd.clone()); let per_turn_config = Self::build_per_turn_config(&session_configuration, cwd.clone()); @@ -705,6 +752,15 @@ impl Session { &per_turn_config.to_models_manager_config(), ) .await; + + let multi_agent_version = match multi_agent_runtime { + TurnMultiAgentRuntime::ResolveAndStore => { + self.resolve_multi_agent_version_for_model(&model_info, &per_turn_config) + } + TurnMultiAgentRuntime::Preview => model_info + .multi_agent_version + .unwrap_or_else(|| per_turn_config.multi_agent_version_from_features()), + }; let plugin_outcome = self .services .plugins_manager @@ -728,6 +784,7 @@ impl Session { &self.services.session_telemetry, session_configuration.provider.clone(), &session_configuration, + multi_agent_version, self.services.user_shell.as_ref(), self.services.shell_zsh_path.as_ref(), self.services.main_execve_wrapper_exe.as_ref(), @@ -781,6 +838,34 @@ impl Session { } pub(crate) async fn new_default_turn_with_sub_id(&self, sub_id: String) -> Arc { + let (session_configuration, turn_environments) = + self.default_turn_configuration_and_environments().await; + self.new_turn_from_configuration( + sub_id, + session_configuration, + /*final_output_json_schema*/ None, + turn_environments, + ) + .await + } + + pub(crate) async fn new_startup_prewarm_turn_with_sub_id( + &self, + sub_id: String, + ) -> Arc { + let (session_configuration, turn_environments) = + self.default_turn_configuration_and_environments().await; + self.new_startup_prewarm_turn_from_configuration( + sub_id, + session_configuration, + turn_environments, + ) + .await + } + + async fn default_turn_configuration_and_environments( + &self, + ) -> (SessionConfiguration, ResolvedTurnEnvironments) { let session_configuration = { let state = self.state.lock().await; state.session_configuration.clone() @@ -797,14 +882,7 @@ impl Session { ResolvedTurnEnvironments::default() } }; - - self.new_turn_from_configuration( - sub_id, - session_configuration, - /*final_output_json_schema*/ None, - turn_environments, - ) - .await + (session_configuration, turn_environments) } fn overlay_runtime_cwd_on_primary_environment( diff --git a/codex-rs/core/src/session_startup_prewarm.rs b/codex-rs/core/src/session_startup_prewarm.rs index 0669e5446..13e0871e9 100644 --- a/codex-rs/core/src/session_startup_prewarm.rs +++ b/codex-rs/core/src/session_startup_prewarm.rs @@ -222,7 +222,7 @@ async fn schedule_startup_prewarm_inner( ) -> CodexResult { let prewarm_started_at = Instant::now(); let startup_turn_context = session - .new_default_turn_with_sub_id(INITIAL_SUBMIT_ID.to_owned()) + .new_startup_prewarm_turn_with_sub_id(INITIAL_SUBMIT_ID.to_owned()) .await; startup_turn_context.session_telemetry.record_startup_phase( "startup_prewarm_create_turn_context", diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 38a8a9627..76d5e6478 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_TOKEN_USAGE_METRIC; use codex_otel::TURN_TOOL_CALL_METRIC; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; @@ -70,11 +71,14 @@ pub(crate) enum InterruptedTurnHistoryMarker { } impl InterruptedTurnHistoryMarker { - pub(crate) fn from_config(config: &Config) -> Self { + pub(crate) fn from_config_and_version( + config: &Config, + multi_agent_version: MultiAgentVersion, + ) -> Self { if !config.agent_interrupt_message_enabled { return Self::Disabled; } - if config.features.enabled(Feature::MultiAgentV2) { + if multi_agent_version == MultiAgentVersion::V2 { Self::Developer } else { Self::ContextualUser @@ -841,7 +845,10 @@ impl Session { if reason == TurnAbortReason::Interrupted && let Some(marker) = interrupted_turn_history_marker( - InterruptedTurnHistoryMarker::from_config(task.turn_context.config.as_ref()), + InterruptedTurnHistoryMarker::from_config_and_version( + task.turn_context.config.as_ref(), + task.turn_context.multi_agent_version, + ), ) { self.record_conversation_items( diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 3624440b2..0b8ebaa14 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -12,6 +12,7 @@ use crate::session::Codex; use crate::session::CodexSpawnArgs; use crate::session::CodexSpawnOk; use crate::session::INITIAL_SUBMIT_ID; +use crate::session::resolve_multi_agent_version; use crate::shell_snapshot::ShellSnapshot; use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; @@ -38,6 +39,7 @@ use codex_protocol::openai_models::ModelPreset; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InitialHistory; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::RolloutItem; @@ -639,10 +641,16 @@ impl ThreadManager { )) })?; let history = stored_thread_to_initial_history(stored_thread, fork_source.rollout_path())?; + let inherited_multi_agent_version = fork_source + .multi_agent_version() + .unwrap_or(MultiAgentVersion::V1); options.initial_history = fork_history_from_snapshot( ForkSnapshot::Interrupted, history, - InterruptedTurnHistoryMarker::from_config(&options.config), + InterruptedTurnHistoryMarker::from_config_and_version( + &options.config, + inherited_multi_agent_version, + ), ); self.start_thread_with_options_and_fork_source(options, Some(forked_from_thread_id)) .await @@ -894,7 +902,18 @@ impl ThreadManager { InitialHistory::Forked(_) => history.forked_from_id(), InitialHistory::New | InitialHistory::Cleared => None, }; - let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config); + let multi_agent_version = self + .state + .effective_multi_agent_version_for_spawn( + &history, + /*session_source*/ None, + /*parent_thread_id*/ None, + forked_from_thread_id, + &config, + ) + .await; + let interrupted_marker = + InterruptedTurnHistoryMarker::from_config_and_version(&config, multi_agent_version); let history = fork_history_from_snapshot(snapshot, history, interrupted_marker); let environments = default_thread_environment_selections( self.state.environment_manager.as_ref(), @@ -1018,6 +1037,52 @@ impl ThreadManagerState { self.threads.write().await.remove(thread_id) } + pub(crate) async fn effective_multi_agent_version_for_spawn( + &self, + initial_history: &InitialHistory, + session_source: Option<&SessionSource>, + parent_thread_id: Option, + forked_from_thread_id: Option, + config: &Config, + ) -> MultiAgentVersion { + self.initial_multi_agent_version_for_spawn( + initial_history, + session_source, + parent_thread_id, + forked_from_thread_id, + ) + .await + .unwrap_or_else(|| config.multi_agent_version_from_features()) + } + + async fn initial_multi_agent_version_for_spawn( + &self, + initial_history: &InitialHistory, + session_source: Option<&SessionSource>, + parent_thread_id: Option, + forked_from_thread_id: Option, + ) -> Option { + let inherited_thread_id = match session_source { + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, .. + })) => Some(*parent_thread_id), + _ => match initial_history { + InitialHistory::Resumed(resumed) => Some(resumed.conversation_id), + InitialHistory::Forked(_) => forked_from_thread_id.or(parent_thread_id), + InitialHistory::New | InitialHistory::Cleared => parent_thread_id, + }, + }; + let inherited_multi_agent_version = match inherited_thread_id { + Some(thread_id) => self + .get_thread(thread_id) + .await + .ok() + .and_then(|thread| thread.multi_agent_version()), + None => None, + }; + resolve_multi_agent_version(initial_history, inherited_multi_agent_version) + } + /// Spawn a new thread with no history using a provided config. pub(crate) async fn spawn_new_thread( &self, @@ -1233,6 +1298,14 @@ impl ThreadManagerState { .parent_rollout_thread_trace_for_source(&session_source, &initial_history) .await; let tracked_session_source = session_source.clone(); + let multi_agent_version = self + .initial_multi_agent_version_for_spawn( + &initial_history, + Some(&session_source), + parent_thread_id, + forked_from_thread_id, + ) + .await; let CodexSpawnOk { codex, thread_id, .. } = Codex::spawn(CodexSpawnArgs { @@ -1262,6 +1335,7 @@ impl ThreadManagerState { analytics_events_client: self.analytics_events_client.clone(), thread_store: Arc::clone(&self.thread_store), attestation_provider: self.attestation_provider.clone(), + inherited_multi_agent_version: multi_agent_version, }) .await?; let new_thread = self diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs index 821b3619c..48ba08743 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ b/codex-rs/core/src/tools/handlers/agent_jobs.rs @@ -1,6 +1,4 @@ use crate::agent::control::SpawnAgentOptions; -use crate::agent::exceeds_thread_spawn_depth_limit; -use crate::agent::next_thread_spawn_depth; use crate::agent::status::is_final; use crate::config::Config; use crate::function_tool::FunctionCallError; @@ -11,6 +9,7 @@ use crate::tools::handlers::parse_arguments; use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; use codex_protocol::protocol::AgentStatus; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::user_input::UserInput; @@ -111,21 +110,22 @@ async fn build_runner_options( turn: &Arc, requested_concurrency: Option, ) -> Result { - let session_source = turn.session_source.clone(); - let child_depth = next_thread_spawn_depth(&session_source); - let max_depth = turn.config.agent_max_depth; - if exceeds_thread_spawn_depth_limit(child_depth, max_depth) { + let multi_agent_version = turn.multi_agent_version; + if multi_agent_version == MultiAgentVersion::Disabled { return Err(FunctionCallError::RespondToModel( - "agent depth limit reached; this session cannot spawn more subagents".to_string(), + "multi-agent runtime is disabled; this session cannot spawn workers".to_string(), )); } - if turn.config.agent_max_threads == Some(0) { + let agent_max_threads = turn + .config + .effective_agent_max_threads(multi_agent_version) + .map_err(|err| FunctionCallError::Fatal(err.to_string()))?; + if agent_max_threads == Some(0) { return Err(FunctionCallError::RespondToModel( "agent thread limit reached; this session cannot spawn more subagents".to_string(), )); } - let max_concurrency = - normalize_concurrency(requested_concurrency, turn.config.agent_max_threads); + let max_concurrency = normalize_concurrency(requested_concurrency, agent_max_threads); let base_instructions = session.get_base_instructions().await; let spawn_config = build_agent_spawn_config(&base_instructions, turn.as_ref())?; Ok(JobRunnerOptions { diff --git a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs index ebe3678eb..78b6aafb9 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs @@ -181,7 +181,7 @@ async fn try_resume_closed_agent( receiver_thread_id: ThreadId, child_depth: i32, ) -> Result<(), FunctionCallError> { - let config = build_agent_resume_config(turn.as_ref(), child_depth)?; + let config = build_agent_resume_config(turn.as_ref())?; Box::pin(session.services.agent_control.resume_agent_from_rollout( config, receiver_thread_id, diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index 56707fb0c..108ddd336 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -116,7 +116,6 @@ async fn handle_spawn_agent( ) .await?; apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?; - apply_spawn_agent_overrides(&mut config, child_depth); let result = Box::pin(session.services.agent_control.spawn_agent_with_metadata( config, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_common.rs b/codex-rs/core/src/tools/handlers/multi_agents_common.rs index f1e7bcd42..8a4edbff3 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_common.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_common.rs @@ -8,7 +8,6 @@ use crate::session::turn_context::TurnContext; use crate::tools::context::FunctionToolOutput; use crate::tools::context::ToolOutput; use crate::tools::context::ToolPayload; -use codex_features::Feature; use codex_models_manager::manager::RefreshStrategy; use codex_protocol::AgentPath; use codex_protocol::ThreadId; @@ -211,12 +210,8 @@ pub(crate) fn build_agent_spawn_config( Ok(config) } -pub(crate) fn build_agent_resume_config( - turn: &TurnContext, - child_depth: i32, -) -> Result { +pub(crate) fn build_agent_resume_config(turn: &TurnContext) -> Result { let mut config = build_agent_shared_config(turn)?; - apply_spawn_agent_overrides(&mut config, child_depth); // For resume, keep base instructions sourced from rollout/session metadata. config.base_instructions = None; Ok(config) @@ -280,13 +275,6 @@ pub(crate) fn apply_spawn_agent_runtime_overrides( Ok(()) } -pub(crate) fn apply_spawn_agent_overrides(config: &mut Config, child_depth: i32) { - if child_depth >= config.agent_max_depth && !config.features.enabled(Feature::MultiAgentV2) { - let _ = config.features.disable(Feature::SpawnCsv); - let _ = config.features.disable(Feature::Collab); - } -} - pub(crate) async fn apply_requested_spawn_agent_model_overrides( session: &Session, turn: &TurnContext, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index e3faf0742..7a27197ea 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -134,6 +134,11 @@ model_reasoning_effort = "minimal" role_name } +fn set_turn_config(turn: &mut TurnContext, config: crate::config::Config) { + turn.multi_agent_version = config.multi_agent_version_from_features(); + turn.config = Arc::new(config); +} + fn expect_text_output(output: T) -> (String, Option) where T: ToolOutput, @@ -382,6 +387,7 @@ async fn multi_agent_v2_spawn_fork_turns_all_rejects_agent_type_override() { .expect("test config should allow feature update"); let turn = TurnContext { config: Arc::new(config), + multi_agent_version: codex_protocol::protocol::MultiAgentVersion::V2, ..turn }; @@ -424,7 +430,7 @@ async fn multi_agent_v2_spawn_defaults_to_full_fork_and_rejects_child_model_over .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let err = SpawnAgentHandlerV2::default() .handle(invocation( @@ -891,7 +897,7 @@ async fn multi_agent_v2_full_history_fork_accepts_explicit_service_tier() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let manager = thread_manager(); let root = manager .start_thread((*turn.config).clone()) @@ -959,6 +965,7 @@ async fn multi_agent_v2_spawn_partial_fork_turns_allows_agent_type_override() { .expect("test config should allow feature update"); let turn = TurnContext { config: Arc::new(config), + multi_agent_version: codex_protocol::protocol::MultiAgentVersion::V2, ..turn }; @@ -1040,7 +1047,7 @@ async fn multi_agent_v2_spawn_requires_task_name() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let invocation = invocation( Arc::new(session), @@ -1074,7 +1081,7 @@ async fn multi_agent_v2_spawn_rejects_legacy_items_field() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let invocation = invocation( Arc::new(session), @@ -1134,7 +1141,7 @@ async fn multi_agent_v2_spawn_returns_path_and_send_message_accepts_relative_pat .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -1231,7 +1238,7 @@ async fn multi_agent_v2_spawn_rejects_legacy_fork_context() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let err = SpawnAgentHandlerV2::default() .handle(invocation( @@ -1271,7 +1278,7 @@ async fn multi_agent_v2_spawn_rejects_invalid_fork_turns_string() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let err = SpawnAgentHandlerV2::default() .handle(invocation( @@ -1311,7 +1318,7 @@ async fn multi_agent_v2_spawn_rejects_zero_fork_turns() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let err = SpawnAgentHandlerV2::default() .handle(invocation( @@ -1340,18 +1347,18 @@ async fn multi_agent_v2_spawn_rejects_zero_fork_turns() { async fn multi_agent_v2_send_message_accepts_root_target_from_child() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); + let mut config = (*turn.config).clone(); + config + .features + .enable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + set_turn_config(&mut turn, config); let root = manager .start_thread((*turn.config).clone()) .await .expect("root thread should start"); session.services.agent_control = manager.agent_control(); session.conversation_id = root.thread_id; - let mut config = (*turn.config).clone(); - config - .features - .enable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - turn.config = Arc::new(config); let child_path = AgentPath::try_from("/root/worker").expect("agent path"); let child_thread_id = session @@ -1416,18 +1423,18 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() { async fn multi_agent_v2_followup_task_rejects_root_target_from_child() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); + let mut config = (*turn.config).clone(); + config + .features + .enable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + set_turn_config(&mut turn, config); let root = manager .start_thread((*turn.config).clone()) .await .expect("root thread should start"); session.services.agent_control = manager.agent_control(); session.conversation_id = root.thread_id; - let mut config = (*turn.config).clone(); - config - .features - .enable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - turn.config = Arc::new(config); let child_path = AgentPath::try_from("/root/worker").expect("agent path"); let child_thread_id = session @@ -1507,7 +1514,7 @@ async fn multi_agent_v2_list_agents_returns_completed_status_and_last_task_messa session.conversation_id = root.thread_id; let mut config = (*turn.config).clone(); let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -1593,15 +1600,15 @@ async fn multi_agent_v2_list_agents_returns_completed_status_and_last_task_messa async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); + let mut config = (*turn.config).clone(); + let _ = config.features.enable(Feature::MultiAgentV2); + set_turn_config(&mut turn, config.clone()); let root = manager .start_thread((*turn.config).clone()) .await .expect("root thread should start"); session.services.agent_control = manager.agent_control(); session.conversation_id = root.thread_id; - let mut config = (*turn.config).clone(); - let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config.clone()); let researcher_path = AgentPath::from_string("/root/researcher".to_string()).expect("path"); let worker_path = AgentPath::from_string("/root/researcher/worker".to_string()).expect("path"); @@ -1688,7 +1695,7 @@ async fn multi_agent_v2_list_agents_omits_closed_agents() { session.conversation_id = root.thread_id; let mut config = (*turn.config).clone(); let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -1752,7 +1759,7 @@ async fn multi_agent_v2_send_message_rejects_legacy_items_field() { session.conversation_id = root.thread_id; let mut config = turn.config.as_ref().clone(); let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -1808,7 +1815,7 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() { session.conversation_id = root.thread_id; let mut config = turn.config.as_ref().clone(); let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -1873,15 +1880,15 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() { async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); + let mut config = turn.config.as_ref().clone(); + let _ = config.features.enable(Feature::MultiAgentV2); + set_turn_config(&mut turn, config); let root = manager .start_thread((*turn.config).clone()) .await .expect("root thread should start"); session.services.agent_control = manager.agent_control(); session.conversation_id = root.thread_id; - let mut config = turn.config.as_ref().clone(); - let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -2016,7 +2023,7 @@ async fn multi_agent_v2_followup_task_rejects_legacy_items_field() { session.conversation_id = root.thread_id; let mut config = turn.config.as_ref().clone(); let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -2069,7 +2076,7 @@ async fn multi_agent_v2_interrupted_turn_does_not_notify_parent() { session.conversation_id = root.thread_id; let mut config = turn.config.as_ref().clone(); let _ = config.features.enable(Feature::MultiAgentV2); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -2149,7 +2156,7 @@ async fn multi_agent_v2_spawn_omits_agent_id_when_named() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let output = SpawnAgentHandlerV2::default() .handle(invocation( @@ -2188,7 +2195,7 @@ async fn multi_agent_v2_spawn_surfaces_task_name_validation_errors() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let invocation = invocation( Arc::new(session), @@ -2398,7 +2405,7 @@ async fn multi_agent_v2_spawn_agent_ignores_configured_max_depth() { .expect("root thread should start"); session.services.agent_control = manager.agent_control(); session.conversation_id = root.thread_id; - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let parent_path = AgentPath::try_from("/root/parent").expect("agent path"); turn.session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, @@ -2871,7 +2878,7 @@ async fn multi_agent_v2_wait_agent_accepts_timeout_only_argument() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -2956,7 +2963,7 @@ async fn multi_agent_v2_wait_agent_rejects_timeout_below_configured_min() { config.multi_agent_v2.min_wait_timeout_ms = 50; config.multi_agent_v2.max_wait_timeout_ms = 1_000; config.multi_agent_v2.default_wait_timeout_ms = 50; - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let Err(err) = WaitAgentHandlerV2::default() .handle(invocation( @@ -2986,7 +2993,7 @@ async fn multi_agent_v2_wait_agent_accepts_explicit_timeout_at_configured_min() config.multi_agent_v2.min_wait_timeout_ms = 1; config.multi_agent_v2.max_wait_timeout_ms = 1_000; config.multi_agent_v2.default_wait_timeout_ms = 50; - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let output = WaitAgentHandlerV2::default() .handle(invocation( @@ -3021,7 +3028,7 @@ async fn multi_agent_v2_wait_agent_uses_configured_default_timeout() { config.multi_agent_v2.min_wait_timeout_ms = 1; config.multi_agent_v2.max_wait_timeout_ms = 1_000; config.multi_agent_v2.default_wait_timeout_ms = 50; - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3076,7 +3083,7 @@ async fn multi_agent_v2_wait_agent_allows_zero_configured_timeout() { config.multi_agent_v2.min_wait_timeout_ms = 0; config.multi_agent_v2.max_wait_timeout_ms = 0; config.multi_agent_v2.default_wait_timeout_ms = 0; - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3116,7 +3123,7 @@ async fn multi_agent_v2_wait_agent_rejects_timeout_above_configured_max() { config.multi_agent_v2.min_wait_timeout_ms = 1; config.multi_agent_v2.max_wait_timeout_ms = 50; config.multi_agent_v2.default_wait_timeout_ms = 1; - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let Err(err) = WaitAgentHandlerV2::default() .handle(invocation( @@ -3146,7 +3153,7 @@ async fn multi_agent_v2_wait_agent_accepts_explicit_timeout_at_configured_max() config.multi_agent_v2.min_wait_timeout_ms = 1; config.multi_agent_v2.max_wait_timeout_ms = 1; config.multi_agent_v2.default_wait_timeout_ms = 1; - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let output = WaitAgentHandlerV2::default() .handle(invocation( @@ -3354,7 +3361,7 @@ async fn multi_agent_v2_wait_agent_returns_summary_for_mailbox_activity() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3448,7 +3455,7 @@ async fn multi_agent_v2_wait_agent_returns_for_already_queued_mail() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3529,7 +3536,7 @@ async fn multi_agent_v2_wait_agent_wakes_on_any_mailbox_notification() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3620,7 +3627,7 @@ async fn multi_agent_v2_wait_agent_does_not_return_completed_content() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3709,7 +3716,7 @@ async fn multi_agent_v2_close_agent_accepts_task_name_target() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3757,7 +3764,7 @@ async fn multi_agent_v2_close_agent_accepts_task_name_target() { async fn multi_agent_v2_close_agent_reaps_stale_task_name_target() { let (mut session, mut turn) = make_session_and_context().await; let mut config = (*turn.config).clone(); - config.agent_max_threads = Some(1); + config.multi_agent_v2.max_concurrent_threads_per_session = 2; config .features .enable(Feature::MultiAgentV2) @@ -3782,7 +3789,7 @@ async fn multi_agent_v2_close_agent_reaps_stale_task_name_target() { .expect("root thread should start"); session.services.agent_control = manager.agent_control(); session.conversation_id = root.thread_id; - turn.config = Arc::new(config.clone()); + set_turn_config(&mut turn, config.clone()); let session = Arc::new(session); let turn = Arc::new(turn); @@ -3888,7 +3895,7 @@ async fn multi_agent_v2_close_agent_rejects_root_target_and_id() { .features .enable(Feature::MultiAgentV2) .expect("test config should allow feature update"); - turn.config = Arc::new(config); + set_turn_config(&mut turn, config); let session = Arc::new(session); let turn = Arc::new(turn); @@ -4283,7 +4290,7 @@ async fn build_agent_resume_config_clears_base_instructions() { .set(AskForApproval::OnRequest) .expect("approval policy set"); - let config = build_agent_resume_config(&turn, /*child_depth*/ 0).expect("resume config"); + let config = build_agent_resume_config(&turn).expect("resume config"); let mut expected = (*turn.config).clone(); expected.base_instructions = None; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index f767f88d7..ed85d329b 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -108,7 +108,6 @@ async fn handle_spawn_agent( ) .await?; apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?; - apply_spawn_agent_overrides(&mut config, child_depth); let spawn_source = thread_spawn_source( session.conversation_id, diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index 56d371407..9d5aa8f8e 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/codex-rs/core/src/tools/spec_plan.rs @@ -1,3 +1,5 @@ +use crate::agent::exceeds_thread_spawn_depth_limit; +use crate::agent::next_thread_spawn_depth; use crate::session::turn_context::TurnContext; use crate::tools::code_mode::execute_spec::create_code_mode_tool; use crate::tools::context::ToolInvocation; @@ -61,6 +63,7 @@ use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::openai_models::ConfigShellToolType; use codex_protocol::openai_models::InputModality; use codex_protocol::openai_models::ToolMode; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_tools::DiscoverableTool; @@ -288,11 +291,18 @@ fn namespace_tools_enabled(turn_context: &TurnContext) -> bool { } fn multi_agent_v2_enabled(turn_context: &TurnContext) -> bool { - turn_context.features.get().enabled(Feature::MultiAgentV2) + turn_context.multi_agent_version == MultiAgentVersion::V2 } fn collab_tools_enabled(turn_context: &TurnContext) -> bool { - multi_agent_v2_enabled(turn_context) || turn_context.features.get().enabled(Feature::Collab) + match turn_context.multi_agent_version { + MultiAgentVersion::Disabled => false, + MultiAgentVersion::V1 => !exceeds_thread_spawn_depth_limit( + next_thread_spawn_depth(&turn_context.session_source), + turn_context.config.agent_max_depth, + ), + MultiAgentVersion::V2 => true, + } } fn goal_tools_enabled(turn_context: &TurnContext) -> bool { @@ -304,7 +314,7 @@ fn goal_tools_enabled(turn_context: &TurnContext) -> bool { } fn agent_jobs_tools_enabled(turn_context: &TurnContext) -> bool { - turn_context.features.get().enabled(Feature::SpawnCsv) + turn_context.features.get().enabled(Feature::SpawnCsv) && collab_tools_enabled(turn_context) } fn agent_jobs_worker_tools_enabled(turn_context: &TurnContext) -> bool { diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index b75e6d2ed..1cb36e474 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -215,6 +215,7 @@ fn set_feature(turn: &mut TurnContext, feature: Feature, enabled: bool) { .disable(feature) .expect("test feature should be disableable in config"); } + turn.multi_agent_version = config.multi_agent_version_from_features(); turn.config = Arc::new(config); turn.tool_mode = turn.model_info.tool_mode.unwrap_or_else(|| { if turn.config.features.enabled(Feature::CodeModeOnly) {