openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-07 20:30:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use resolved multi-agent version across runtime

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-05-29 10:19:34 -07:00
parent abca78f62e
commit 7e7a9eb340
14 changed files with 388 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+84 -15
View File
@@ -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<String>,
pub(crate) fork_mode: Option<SpawnAgentForkMode>,
pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
pub(crate) parent_multi_agent_version: Option<MultiAgentVersion>,
pub(crate) child_multi_agent_version: Option<MultiAgentVersion>,
}
#[derive(Clone, Debug)]
@@ -218,7 +221,35 @@ impl AgentControl {
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
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<String> =
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<MultiAgentVersion>,
) -> CodexResult<ThreadId> {
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<MultiAgentVersion>,
) -> CodexResult<ThreadId> {
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;
};
codex-rs/core/src/agent/control_tests.rs
+81 -10
View File
@@ -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");
codex-rs/core/src/session/mod.rs
+18 -9
View File
@@ -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()
codex-rs/core/src/session/tests.rs
+42
View File
@@ -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;
codex-rs/core/src/tasks/mod.rs
+17 -5
View File
@@ -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<MultiAgentVersion>,
) -> 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(),
codex-rs/core/src/thread_manager.rs
+6 -2
View File
@@ -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(),
codex-rs/core/src/thread_manager_tests.rs
+18
View File
@@ -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 = [
codex-rs/core/src/tools/handlers/agent_jobs.rs
+17 -3
View File
@@ -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<MultiAgentVersion>,
}
#[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()
},
)
codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs
+1
View File
@@ -192,6 +192,7 @@ async fn try_resume_closed_agent(
/*agent_role*/ None,
/*task_name*/ None,
)?,
turn.multi_agent_version,
))
.await
.map(|_| ())
codex-rs/core/src/tools/handlers/multi_agents/spawn.rs
+5 -1
View File
@@ -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
codex-rs/core/src/tools/handlers/multi_agents_common.rs
+33 -3
View File
@@ -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<Config, FunctionCallError> {
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<MultiAgentVersion>,
) {
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<Option<MultiAgentVersion>, 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,
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+5 -1
View File
@@ -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,
},
),
)
codex-rs/core/src/tools/spec_plan.rs
+15
View File
@@ -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 {
codex-rs/core/src/tools/spec_plan_tests.rs
+46
View File
@@ -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]