From 7e7a9eb3406dd73c6d04455fd53fc9af7769cd2b Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Fri, 29 May 2026 10:19:34 -0700 Subject: [PATCH] Use resolved multi-agent version across runtime --- codex-rs/core/src/agent/control.rs | 99 ++++++++++++++++--- codex-rs/core/src/agent/control_tests.rs | 91 +++++++++++++++-- codex-rs/core/src/session/mod.rs | 27 +++-- codex-rs/core/src/session/tests.rs | 42 ++++++++ codex-rs/core/src/tasks/mod.rs | 22 ++++- codex-rs/core/src/thread_manager.rs | 8 +- codex-rs/core/src/thread_manager_tests.rs | 18 ++++ .../core/src/tools/handlers/agent_jobs.rs | 20 +++- .../handlers/multi_agents/resume_agent.rs | 1 + .../src/tools/handlers/multi_agents/spawn.rs | 6 +- .../src/tools/handlers/multi_agents_common.rs | 36 ++++++- .../tools/handlers/multi_agents_v2/spawn.rs | 6 +- codex-rs/core/src/tools/spec_plan.rs | 15 +++ codex-rs/core/src/tools/spec_plan_tests.rs | 46 +++++++++ 14 files changed, 388 insertions(+), 49 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 7e87bf73d6..203d13a71e 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -21,6 +21,7 @@ use codex_protocol::error::Result as CodexResult; use codex_protocol::models::ContentItem; use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::MultiAgentVersion; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::Op; @@ -55,6 +56,8 @@ pub(crate) struct SpawnAgentOptions { pub(crate) fork_parent_spawn_call_id: Option, pub(crate) fork_mode: Option, pub(crate) environments: Option>, + pub(crate) parent_multi_agent_version: Option, + pub(crate) child_multi_agent_version: Option, } #[derive(Clone, Debug)] @@ -218,7 +221,35 @@ impl AgentControl { options: SpawnAgentOptions, ) -> CodexResult { let state = self.upgrade()?; - let mut reservation = self.state.reserve_spawn_slot(config.agent_max_threads)?; + let parent_multi_agent_version = options.parent_multi_agent_version.or_else(|| { + if config.features.enabled(Feature::MultiAgentV2) { + Some(MultiAgentVersion::V2) + } else if config.features.enabled(Feature::Collab) { + Some(MultiAgentVersion::V1) + } else { + None + } + }); + let child_multi_agent_version = options.child_multi_agent_version.or_else(|| { + if config.features.enabled(Feature::MultiAgentV2) { + Some(MultiAgentVersion::V2) + } else if config.features.enabled(Feature::Collab) { + Some(MultiAgentVersion::V1) + } else { + None + } + }); + let max_threads = if parent_multi_agent_version == Some(MultiAgentVersion::V2) { + Some( + config + .multi_agent_v2 + .max_concurrent_threads_per_session + .saturating_sub(1), + ) + } else { + config.agent_max_threads + }; + let mut reservation = self.state.reserve_spawn_slot(max_threads)?; let inherited_shell_snapshot = self .inherited_shell_snapshot_for_source(&state, session_source.as_ref()) .await; @@ -339,7 +370,9 @@ impl AgentControl { self.send_input(new_thread.thread_id, initial_operation) .await?; - if !new_thread.thread.enabled(Feature::MultiAgentV2) { + if child_multi_agent_version != Some(MultiAgentVersion::V2) + || agent_metadata.agent_path.is_none() + { let child_reference = agent_metadata .agent_path .as_ref() @@ -416,9 +449,27 @@ impl AgentControl { forked_rollout_items = truncate_rollout_to_last_n_fork_turns(&forked_rollout_items, *last_n_turns); } + let parent_multi_agent_version = options.parent_multi_agent_version.or_else(|| { + if config.features.enabled(Feature::MultiAgentV2) { + Some(MultiAgentVersion::V2) + } else if config.features.enabled(Feature::Collab) { + Some(MultiAgentVersion::V1) + } else { + None + } + }); + let child_multi_agent_version = options.child_multi_agent_version.or_else(|| { + if config.features.enabled(Feature::MultiAgentV2) { + Some(MultiAgentVersion::V2) + } else if config.features.enabled(Feature::Collab) { + Some(MultiAgentVersion::V1) + } else { + None + } + }); 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 parent_multi_agent_version == Some(MultiAgentVersion::V2) { let parent_config = parent_thread.codex.session.get_config().await; [ parent_config @@ -436,7 +487,7 @@ impl AgentControl { } else { Vec::new() } - } else if config.features.enabled(Feature::MultiAgentV2) { + } else if parent_multi_agent_version == Some(MultiAgentVersion::V2) { [ config.multi_agent_v2.root_agent_usage_hint_text.clone(), config.multi_agent_v2.subagent_usage_hint_text.clone(), @@ -472,7 +523,7 @@ impl AgentControl { } } if preserve_reference_context_item - && config.features.enabled(Feature::MultiAgentV2) + && child_multi_agent_version == Some(MultiAgentVersion::V2) && let Some(subagent_usage_hint_text) = config.multi_agent_v2.subagent_usage_hint_text.clone() && let Some(subagent_usage_hint_message) = @@ -505,12 +556,14 @@ impl AgentControl { config: crate::config::Config, thread_id: ThreadId, session_source: SessionSource, + multi_agent_version: Option, ) -> CodexResult { let root_depth = thread_spawn_depth(&session_source).unwrap_or(0); let resumed_thread_id = Box::pin(self.resume_single_agent_from_rollout( config.clone(), thread_id, session_source, + multi_agent_version, )) .await?; let state = self.upgrade()?; @@ -556,6 +609,7 @@ impl AgentControl { config.clone(), child_thread_id, child_session_source, + multi_agent_version, )) .await { @@ -580,17 +634,37 @@ impl AgentControl { mut config: crate::config::Config, thread_id: ThreadId, session_source: SessionSource, + multi_agent_version: Option, ) -> CodexResult { + let multi_agent_version = multi_agent_version.or_else(|| { + if config.features.enabled(Feature::MultiAgentV2) { + Some(MultiAgentVersion::V2) + } else if config.features.enabled(Feature::Collab) { + Some(MultiAgentVersion::V1) + } else { + None + } + }); if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = &session_source && *depth >= config.agent_max_depth - && !config.features.enabled(Feature::MultiAgentV2) + && multi_agent_version != Some(MultiAgentVersion::V2) { 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 max_threads = if multi_agent_version == Some(MultiAgentVersion::V2) { + Some( + config + .multi_agent_v2 + .max_concurrent_threads_per_session + .saturating_sub(1), + ) + } else { + config.agent_max_threads + }; + let mut reservation = self.state.reserve_spawn_slot(max_threads)?; let (session_source, agent_metadata) = match session_source { SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, @@ -659,7 +733,8 @@ 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 != Some(MultiAgentVersion::V2) || agent_metadata.agent_path.is_none() + { let child_reference = agent_metadata .agent_path .as_ref() @@ -1042,14 +1117,8 @@ impl AgentControl { let Ok(state) = control.upgrade() else { return; }; - 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) - { + if child_agent_path.is_some() { let Some(child_agent_path) = child_agent_path.clone() else { return; }; diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 43a6e47bfd..edcf6d0574 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -17,6 +17,7 @@ use codex_protocol::config_types::ModeKind; use codex_protocol::models::ContentItem; use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::MultiAgentVersion; use codex_protocol::protocol::CompactedItem; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; @@ -341,7 +342,12 @@ async fn resume_agent_errors_when_manager_dropped() { let control = AgentControl::default(); let (_home, config) = test_config().await; let err = control - .resume_agent_from_rollout(config, ThreadId::new(), SessionSource::Exec) + .resume_agent_from_rollout( + config, + ThreadId::new(), + SessionSource::Exec, + /*multi_agent_version*/ None, + ) .await .expect_err("resume_agent should fail without a manager"); assert_eq!( @@ -561,13 +567,11 @@ async fn spawn_agent_creates_thread_and_sends_prompt() { async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { let harness = AgentControlHarness::new().await; let mut parent_config = harness.config.clone(); - let _ = parent_config.features.enable(Feature::MultiAgentV2); parent_config.multi_agent_v2.root_agent_usage_hint_text = Some("Parent root guidance.".to_string()); parent_config.multi_agent_v2.subagent_usage_hint_text = Some("Parent subagent guidance.".to_string()); let mut child_config = harness.config.clone(); - let _ = child_config.features.enable(Feature::MultiAgentV2); child_config.multi_agent_v2.root_agent_usage_hint_text = Some("Child root guidance.".to_string()); child_config.multi_agent_v2.subagent_usage_hint_text = @@ -662,6 +666,8 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), fork_mode: Some(SpawnAgentForkMode::FullHistory), + parent_multi_agent_version: Some(MultiAgentVersion::V2), + child_multi_agent_version: Some(MultiAgentVersion::V2), ..Default::default() }, ) @@ -1314,6 +1320,52 @@ async fn spawn_agent_respects_max_threads_limit() { .expect("shutdown agent"); } +#[tokio::test] +async fn spawn_agent_uses_multi_agent_v2_selector_thread_limit() { + let harness = AgentControlHarness::new().await; + let mut config = harness.config.clone(); + config.agent_max_threads = Some(4); + config.multi_agent_v2.max_concurrent_threads_per_session = 2; + let options = SpawnAgentOptions { + parent_multi_agent_version: Some(MultiAgentVersion::V2), + child_multi_agent_version: Some(MultiAgentVersion::V2), + ..Default::default() + }; + + let first_agent_id = harness + .control + .spawn_agent_with_metadata( + config.clone(), + text_input("hello"), + /*session_source*/ None, + options.clone(), + ) + .await + .expect("spawn_agent should succeed") + .thread_id; + + let err = harness + .control + .spawn_agent_with_metadata( + config, + text_input("hello again"), + /*session_source*/ None, + options, + ) + .await + .expect_err("spawn_agent should use the v2 thread limit"); + let CodexErr::AgentLimitReached { max_threads } = err else { + panic!("expected CodexErr::AgentLimitReached"); + }; + assert_eq!(max_threads, 1); + + let _ = harness + .control + .shutdown_live_agent(first_agent_id) + .await + .expect("shutdown agent"); +} + #[tokio::test] async fn spawn_agent_releases_slot_after_shutdown() { let max_threads = 1usize; @@ -1441,7 +1493,12 @@ async fn resume_agent_respects_max_threads_limit() { .expect("spawn_agent should succeed for active slot"); let err = control - .resume_agent_from_rollout(config, resumable_id, SessionSource::Exec) + .resume_agent_from_rollout( + config, + resumable_id, + SessionSource::Exec, + /*multi_agent_version*/ None, + ) .await .expect_err("resume should respect max threads"); let CodexErr::AgentLimitReached { @@ -1475,7 +1532,12 @@ async fn resume_agent_releases_slot_after_resume_failure() { let control = manager.agent_control(); let _ = control - .resume_agent_from_rollout(config.clone(), ThreadId::new(), SessionSource::Exec) + .resume_agent_from_rollout( + config.clone(), + ThreadId::new(), + SessionSource::Exec, + /*multi_agent_version*/ None, + ) .await .expect_err("resume should fail for missing rollout path"); @@ -1628,15 +1690,13 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() { } #[tokio::test] -async fn multi_agent_v2_completion_queues_message_for_direct_parent() { +async fn completion_watcher_queues_message_for_direct_parent() { let harness = AgentControlHarness::new().await; let (_root_thread_id, root_thread) = harness.start_thread().await; let (worker_thread_id, _worker_thread) = harness.start_thread().await; - let mut tester_config = harness.config.clone(); - let _ = tester_config.features.enable(Feature::MultiAgentV2); let tester_thread_id = harness .manager - .start_thread(tester_config.clone()) + .start_thread(harness.config.clone()) .await .expect("tester thread should start") .thread_id; @@ -1964,6 +2024,7 @@ async fn resume_thread_subagent_restores_stored_nickname_and_role() { agent_nickname: None, agent_role: None, }), + /*multi_agent_version*/ None, ) .await .expect("resume should succeed"); @@ -2037,7 +2098,12 @@ async fn resume_agent_from_rollout_reads_archived_rollout_path() { let resumed_thread_id = harness .control - .resume_agent_from_rollout(harness.config.clone(), child_thread_id, SessionSource::Exec) + .resume_agent_from_rollout( + harness.config.clone(), + child_thread_id, + SessionSource::Exec, + /*multi_agent_version*/ None, + ) .await .expect("resume should find archived rollout"); assert_eq!(resumed_thread_id, child_thread_id); @@ -2480,6 +2546,7 @@ async fn resume_agent_from_rollout_does_not_reopen_closed_descendants() { harness.config.clone(), parent_thread_id, SessionSource::Exec, + /*multi_agent_version*/ None, ) .await .expect("single-thread resume should succeed"); @@ -2576,6 +2643,7 @@ async fn resume_closed_child_reopens_open_descendants() { agent_nickname: None, agent_role: None, }), + /*multi_agent_version*/ None, ) .await .expect("child resume should succeed"); @@ -2668,6 +2736,7 @@ async fn resume_agent_from_rollout_reopens_open_descendants_after_manager_shutdo harness.config.clone(), parent_thread_id, SessionSource::Exec, + /*multi_agent_version*/ None, ) .await .expect("tree resume should succeed"); @@ -2781,6 +2850,7 @@ async fn resume_agent_from_rollout_uses_edge_data_when_descendant_metadata_sourc harness.config.clone(), parent_thread_id, SessionSource::Exec, + /*multi_agent_version*/ None, ) .await .expect("tree resume should succeed"); @@ -2896,6 +2966,7 @@ async fn resume_agent_from_rollout_skips_descendants_when_parent_resume_fails() harness.config.clone(), parent_thread_id, SessionSource::Exec, + /*multi_agent_version*/ None, ) .await .expect("root resume should succeed"); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 3ab9e72731..a1eda42041 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -486,14 +486,6 @@ impl Codex { 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) @@ -519,7 +511,7 @@ impl Codex { ) }; - let config = Arc::new(config); + let mut config = Arc::new(config); let refresh_strategy = if session_source.is_non_root_agent() { codex_models_manager::manager::RefreshStrategy::Offline } else { @@ -544,6 +536,23 @@ impl Codex { let model_info = models_manager .get_model_info(model.as_str(), &config.to_models_manager_config()) .await; + let multi_agent_version = model_info.multi_agent_version.or_else(|| { + if config.features.enabled(Feature::MultiAgentV2) { + Some(MultiAgentVersion::V2) + } else if config.features.enabled(Feature::Collab) { + Some(MultiAgentVersion::V1) + } else { + None + } + }); + if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = &session_source + && *depth >= config.agent_max_depth + && multi_agent_version != Some(MultiAgentVersion::V2) + { + let config = Arc::make_mut(&mut config); + let _ = config.features.disable(Feature::SpawnCsv); + let _ = config.features.disable(Feature::Collab); + } let base_instructions = config .base_instructions .clone() diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index aa2586cae4..14fcf0fbd7 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -44,6 +44,7 @@ use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::PermissionProfile; use codex_protocol::models::SandboxEnforcement; use codex_protocol::openai_models::ModelServiceTier; +use codex_protocol::openai_models::MultiAgentVersion; use codex_protocol::permissions::FileSystemAccessMode; use codex_protocol::permissions::FileSystemPath; use codex_protocol::permissions::FileSystemSandboxEntry; @@ -7172,6 +7173,47 @@ async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_feature_dis ); } +#[tokio::test] +async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_selector_is_v1() { + let (session, mut turn_context) = + make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ true).await; + Arc::get_mut(&mut turn_context) + .expect("turn context should not be shared") + .multi_agent_version = Some(MultiAgentVersion::V1); + + let initial_context = session.build_initial_context(turn_context.as_ref()).await; + + let developer_messages = developer_message_texts(&initial_context); + assert!( + !developer_messages.iter().any(|message| { + matches!( + message.as_slice(), + ["Root guidance."] | ["Subagent guidance."] + ) + }), + "did not expect multi-agent v2 usage hint developer messages, got {developer_messages:?}" + ); +} + +#[tokio::test] +async fn build_initial_context_adds_multi_agent_v2_usage_hint_when_selector_is_v2() { + let (session, mut turn_context) = + make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ false).await; + Arc::get_mut(&mut turn_context) + .expect("turn context should not be shared") + .multi_agent_version = Some(MultiAgentVersion::V2); + + let initial_context = session.build_initial_context(turn_context.as_ref()).await; + + let developer_messages = developer_message_texts(&initial_context); + assert!( + developer_messages + .iter() + .any(|message| message.as_slice() == ["Root guidance."]), + "expected standalone root usage hint developer message, got {developer_messages:?}" + ); +} + #[tokio::test] async fn build_initial_context_omits_default_image_save_location_with_image_history() { let (session, turn_context) = make_session_and_context().await; diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 38a8a96276..4c43d52c64 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -52,6 +52,7 @@ use codex_protocol::protocol::WarningEvent; use codex_features::Feature; use codex_protocol::models::ContentItem; +use codex_protocol::openai_models::MultiAgentVersion; pub(crate) use compact::CompactTask; pub(crate) use regular::RegularTask; pub(crate) use review::ReviewTask; @@ -70,11 +71,20 @@ pub(crate) enum InterruptedTurnHistoryMarker { } impl InterruptedTurnHistoryMarker { - pub(crate) fn from_config(config: &Config) -> Self { + pub(crate) fn from_config( + config: &Config, + multi_agent_version: Option, + ) -> Self { if !config.agent_interrupt_message_enabled { return Self::Disabled; } - if config.features.enabled(Feature::MultiAgentV2) { + if multi_agent_version.or_else(|| { + config + .features + .enabled(Feature::MultiAgentV2) + .then_some(MultiAgentVersion::V2) + }) == Some(MultiAgentVersion::V2) + { Self::Developer } else { Self::ContextualUser @@ -840,9 +850,11 @@ impl Session { .await; if reason == TurnAbortReason::Interrupted - && let Some(marker) = interrupted_turn_history_marker( - InterruptedTurnHistoryMarker::from_config(task.turn_context.config.as_ref()), - ) + && let Some(marker) = + interrupted_turn_history_marker(InterruptedTurnHistoryMarker::from_config( + task.turn_context.config.as_ref(), + task.turn_context.multi_agent_version, + )) { self.record_conversation_items( task.turn_context.as_ref(), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 3eec8315b1..d7c0fccdd4 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -640,7 +640,10 @@ impl ThreadManager { options.initial_history = fork_history_from_snapshot( ForkSnapshot::Interrupted, history, - InterruptedTurnHistoryMarker::from_config(&options.config), + InterruptedTurnHistoryMarker::from_config( + &options.config, + /*multi_agent_version*/ None, + ), ); self.start_thread_with_options_and_fork_source(options, Some(forked_from_thread_id)) .await @@ -895,7 +898,8 @@ impl ThreadManager { InitialHistory::Forked(_) => history.forked_from_id(), InitialHistory::New | InitialHistory::Cleared => None, }; - let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config); + let interrupted_marker = + InterruptedTurnHistoryMarker::from_config(&config, /*multi_agent_version*/ None); let history = fork_history_from_snapshot(snapshot, history, interrupted_marker); let environments = default_thread_environment_selections( self.state.environment_manager.as_ref(), diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 76880eeb9e..cf8bc1221b 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -14,6 +14,7 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ReasoningItemReasoningSummary; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ModelsResponse; +use codex_protocol::openai_models::MultiAgentVersion; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InternalSessionSource; @@ -64,6 +65,23 @@ fn developer_interrupted_marker() -> ResponseItem { .expect("developer interrupted marker should be enabled") } +#[tokio::test] +async fn interrupted_turn_history_marker_uses_resolved_multi_agent_version() { + let mut config = test_config().await; + let _ = config.features.enable(Feature::MultiAgentV2); + + assert_eq!( + ( + InterruptedTurnHistoryMarker::from_config(&config, Some(MultiAgentVersion::V1),), + InterruptedTurnHistoryMarker::from_config(&config, Some(MultiAgentVersion::V2),), + ), + ( + InterruptedTurnHistoryMarker::ContextualUser, + InterruptedTurnHistoryMarker::Developer, + ) + ); +} + #[test] fn truncates_before_requested_user_message() { let items = [ diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs index 4c3cd34c2e..f8a72739f6 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ b/codex-rs/core/src/tools/handlers/agent_jobs.rs @@ -10,6 +10,7 @@ use crate::tools::handlers::multi_agents::build_agent_spawn_config; use crate::tools::handlers::parse_arguments; use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; +use codex_protocol::openai_models::MultiAgentVersion; use codex_protocol::protocol::AgentStatus; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; @@ -89,6 +90,7 @@ struct ReportAgentJobResultToolResult { struct JobRunnerOptions { max_concurrency: usize, spawn_config: Config, + multi_agent_version: Option, } #[derive(Debug, Clone)] @@ -119,18 +121,28 @@ async fn build_runner_options( "agent depth limit reached; this session cannot spawn more subagents".to_string(), )); } - if turn.config.agent_max_threads == Some(0) { + let max_threads = if turn.multi_agent_version == Some(MultiAgentVersion::V2) { + Some( + turn.config + .multi_agent_v2 + .max_concurrent_threads_per_session + .saturating_sub(1), + ) + } else { + turn.config.agent_max_threads + }; + if 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, max_threads); let base_instructions = session.get_base_instructions().await; let spawn_config = build_agent_spawn_config(&base_instructions, turn.as_ref())?; Ok(JobRunnerOptions { max_concurrency, spawn_config, + multi_agent_version: turn.multi_agent_version, }) } @@ -212,6 +224,8 @@ async fn run_agent_job_loop( )))), SpawnAgentOptions { environments: Some(turn.environments.to_selections()), + parent_multi_agent_version: options.multi_agent_version, + child_multi_agent_version: options.multi_agent_version, ..Default::default() }, ) 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 a536c8d704..97948b6ec5 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 @@ -192,6 +192,7 @@ async fn try_resume_closed_agent( /*agent_role*/ None, /*task_name*/ None, )?, + turn.multi_agent_version, )) .await .map(|_| ()) 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 a6461b3581..aa9de567dc 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -109,7 +109,9 @@ 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 child_multi_agent_version = + resolve_multi_agent_version_for_config(&session, &config).await?; + apply_spawn_agent_overrides(&mut config, child_depth, child_multi_agent_version); let result = Box::pin(session.services.agent_control.spawn_agent_with_metadata( config, @@ -125,6 +127,8 @@ async fn handle_spawn_agent( fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()), fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), environments: Some(turn.environments.to_selections()), + parent_multi_agent_version: turn.multi_agent_version, + child_multi_agent_version, }, )) .await 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 f1e7bcd429..a4818f9fdd 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_common.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_common.rs @@ -15,6 +15,7 @@ use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; use codex_protocol::models::BaseInstructions; use codex_protocol::models::ResponseInputItem; +use codex_protocol::openai_models::MultiAgentVersion; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::openai_models::ReasoningEffortPreset; use codex_protocol::protocol::CollabAgentRef; @@ -216,7 +217,7 @@ pub(crate) fn build_agent_resume_config( child_depth: i32, ) -> Result { let mut config = build_agent_shared_config(turn)?; - apply_spawn_agent_overrides(&mut config, child_depth); + apply_spawn_agent_overrides(&mut config, child_depth, turn.multi_agent_version); // For resume, keep base instructions sourced from rollout/session metadata. config.base_instructions = None; Ok(config) @@ -280,13 +281,42 @@ 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) { +pub(crate) fn apply_spawn_agent_overrides( + config: &mut Config, + child_depth: i32, + multi_agent_version: Option, +) { + if child_depth >= config.agent_max_depth && multi_agent_version != Some(MultiAgentVersion::V2) { let _ = config.features.disable(Feature::SpawnCsv); let _ = config.features.disable(Feature::Collab); } } +pub(crate) async fn resolve_multi_agent_version_for_config( + session: &Session, + config: &Config, +) -> Result, FunctionCallError> { + let model = config.model.as_deref().ok_or_else(|| { + FunctionCallError::RespondToModel( + "spawn_agent could not resolve the child model runtime".to_string(), + ) + })?; + let model_info = session + .services + .models_manager + .get_model_info(model, &config.to_models_manager_config()) + .await; + Ok(model_info.multi_agent_version.or_else(|| { + if config.features.enabled(Feature::MultiAgentV2) { + Some(MultiAgentVersion::V2) + } else if config.features.enabled(Feature::Collab) { + Some(MultiAgentVersion::V1) + } else { + None + } + })) +} + 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_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index 9a4e39195d..2a227722df 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,9 @@ 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 child_multi_agent_version = + resolve_multi_agent_version_for_config(&session, &config).await?; + apply_spawn_agent_overrides(&mut config, child_depth, child_multi_agent_version); let spawn_source = thread_spawn_source( session.conversation_id, @@ -145,6 +147,8 @@ async fn handle_spawn_agent( fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()), fork_mode, environments: Some(turn.environments.to_selections()), + parent_multi_agent_version: turn.multi_agent_version, + child_multi_agent_version, }, ), ) diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index 35f937dd80..8bc64b09ee 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/codex-rs/core/src/tools/spec_plan.rs @@ -1,3 +1,4 @@ +use crate::guardian::is_guardian_reviewer_source; use crate::session::turn_context::TurnContext; use crate::tools::code_mode::execute_spec::create_code_mode_tool; use crate::tools::context::ToolInvocation; @@ -639,6 +640,20 @@ fn add_core_utility_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut fn add_collaboration_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) { let turn_context = context.turn_context; + if is_guardian_reviewer_source(&turn_context.session_source) + || matches!( + &turn_context.session_source, + SessionSource::SubAgent(SubAgentSource::Review) + ) + || matches!( + &turn_context.session_source, + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) + if turn_context.multi_agent_version == Some(MultiAgentVersion::V1) + && *depth >= turn_context.config.agent_max_depth + ) + { + return; + } if turn_context.multi_agent_version.is_some() { if turn_context.multi_agent_version == Some(MultiAgentVersion::V2) { let exposure = if turn_context.config.multi_agent_v2.non_code_mode_only { diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index 918c91ece0..6ebaa439dc 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -8,6 +8,7 @@ use codex_mcp::ToolInfo; use codex_model_provider::create_model_provider; use codex_model_provider_info::AMAZON_BEDROCK_PROVIDER_ID; use codex_model_provider_info::ModelProviderInfo; +use codex_protocol::ThreadId; use codex_protocol::config_types::WebSearchMode; use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::openai_models::ApplyPatchToolType; @@ -31,6 +32,7 @@ use codex_tools::ToolSpec; use pretty_assertions::assert_eq; use serde_json::json; +use crate::guardian::GUARDIAN_REVIEWER_NAME; use crate::session::tests::make_session_and_context; use crate::session::turn_context::TurnContext; use crate::tools::handlers::multi_agents_spec::MULTI_AGENT_V1_NAMESPACE; @@ -878,6 +880,50 @@ async fn multi_agent_version_selector_overrides_feature_flags() { "close_agent", "list_agents", ]); + + let review = probe(|turn| { + turn.multi_agent_version = Some(MultiAgentVersion::V2); + turn.session_source = SessionSource::SubAgent(SubAgentSource::Review); + }) + .await; + review.assert_visible_lacks(&[ + MULTI_AGENT_V1_NAMESPACE, + "spawn_agent", + "send_message", + "followup_task", + "wait_agent", + "close_agent", + "list_agents", + ]); + + let guardian_review = probe(|turn| { + turn.multi_agent_version = Some(MultiAgentVersion::V2); + turn.session_source = + SessionSource::SubAgent(SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string())); + }) + .await; + guardian_review.assert_visible_lacks(&[ + MULTI_AGENT_V1_NAMESPACE, + "spawn_agent", + "send_message", + "followup_task", + "wait_agent", + "close_agent", + "list_agents", + ]); + + let max_depth_v1 = probe(|turn| { + turn.multi_agent_version = Some(MultiAgentVersion::V1); + turn.session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: ThreadId::new(), + depth: turn.config.agent_max_depth, + agent_path: None, + agent_nickname: None, + agent_role: None, + }); + }) + .await; + max_depth_v1.assert_visible_lacks(&[MULTI_AGENT_V1_NAMESPACE]); } #[tokio::test]