diff --git a/codex-rs/app-server/tests/common/rollout.rs b/codex-rs/app-server/tests/common/rollout.rs index 408ef646fd..6d446e1c42 100644 --- a/codex-rs/app-server/tests/common/rollout.rs +++ b/codex-rs/app-server/tests/common/rollout.rs @@ -2,7 +2,6 @@ use anyhow::Result; use codex_protocol::ThreadId; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GitInfo; -use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; @@ -195,7 +194,7 @@ fn create_fake_rollout_with_source_and_parent_thread_id( base_instructions: None, dynamic_tools: None, memory_mode: None, - multi_agent_version: Some(MultiAgentVersion::V1), + multi_agent_version: None, }; let payload = serde_json::to_value(SessionMetaLine { meta, @@ -281,7 +280,7 @@ pub fn create_fake_rollout_with_text_elements( base_instructions: None, dynamic_tools: None, memory_mode: None, - multi_agent_version: Some(MultiAgentVersion::V1), + multi_agent_version: None, }; let payload = serde_json::to_value(SessionMetaLine { meta, diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 306c0d79f4..c7a9f8db5c 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -201,7 +201,6 @@ impl AgentControl { .get_thread(*parent_thread_id) .await? .multi_agent_version() - .await } else { config.multi_agent_version_from_features() } @@ -374,7 +373,7 @@ impl AgentControl { self.send_input(new_thread.thread_id, initial_operation) .await?; - if new_thread.thread.multi_agent_version().await != Some(MultiAgentVersion::V2) { + if new_thread.thread.multi_agent_version() != Some(MultiAgentVersion::V2) { let child_reference = agent_metadata .agent_path .as_ref() @@ -631,7 +630,7 @@ impl AgentControl { }) = &session_source { match state.get_thread(*parent_thread_id).await { - Ok(parent_thread) => parent_thread.multi_agent_version().await, + Ok(parent_thread) => parent_thread.multi_agent_version(), Err(_) => None, } } else { @@ -721,7 +720,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.multi_agent_version().await != Some(MultiAgentVersion::V2) { + if resumed_thread.thread.multi_agent_version() != Some(MultiAgentVersion::V2) { let child_reference = agent_metadata .agent_path .as_ref() @@ -1108,7 +1107,7 @@ impl AgentControl { let message = format_subagent_notification_message(child_reference.as_str(), &status); let child_uses_multi_agent_v2 = match child_thread.as_ref() { Some(child_thread) => { - child_thread.multi_agent_version().await == Some(MultiAgentVersion::V2) + child_thread.multi_agent_version() == Some(MultiAgentVersion::V2) } None => true, }; diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 171012257d..2d00fc2845 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -611,10 +611,7 @@ async fn spawn_agent_inherits_parent_multi_agent_version_over_child_feature_flag .await .expect("child thread should be registered"); - assert_eq!( - child.multi_agent_version().await, - Some(MultiAgentVersion::V2) - ); + assert_eq!(child.multi_agent_version(), Some(MultiAgentVersion::V2)); } #[tokio::test] @@ -721,7 +718,6 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { })), parent_thread .multi_agent_version() - .await .expect("parent thread should have a multi-agent version"), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), @@ -936,7 +932,6 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() { })), parent_thread .multi_agent_version() - .await .expect("parent thread should have a multi-agent version"), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id), @@ -1010,7 +1005,6 @@ async fn spawn_agent_fork_flushes_parent_rollout_before_loading_history() { })), parent_thread .multi_agent_version() - .await .expect("parent thread should have a multi-agent version"), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), @@ -1131,7 +1125,6 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() { })), parent_thread .multi_agent_version() - .await .expect("parent thread should have a multi-agent version"), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), @@ -1246,7 +1239,6 @@ async fn spawn_agent_fork_last_n_turns_drops_parent_startup_prefix_when_under_li })), parent_thread .multi_agent_version() - .await .expect("parent thread should have a multi-agent version"), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id), @@ -1360,7 +1352,6 @@ async fn spawn_agent_fork_last_n_turns_strips_parent_usage_hints() { })), parent_thread .multi_agent_version() - .await .expect("parent thread should have a multi-agent version"), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id), diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index d3e35187af..d39d4587eb 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -503,8 +503,8 @@ impl CodexThread { self.codex.session.get_config().await } - pub async fn multi_agent_version(&self) -> Option { - self.codex.session.multi_agent_version().await + pub fn multi_agent_version(&self) -> Option { + self.codex.session.multi_agent_version() } pub(crate) fn set_multi_agent_version_if_unset( diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index ac0bad9230..6e7bba2afc 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -9912,40 +9912,6 @@ subagent_usage_hint_text = "" Ok(()) } -#[tokio::test] -async fn multi_agent_version_resolution_prefers_v2_then_v1_then_disabled() -> std::io::Result<()> { - let mut config = test_config().await; - config - .features - .disable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - config - .features - .disable(Feature::Collab) - .expect("test config should allow feature update"); - assert_eq!(config.multi_agent_version_from_features(), None); - - config - .features - .enable(Feature::Collab) - .expect("test config should allow feature update"); - assert_eq!( - config.multi_agent_version_from_features(), - Some(MultiAgentVersion::V1) - ); - - config - .features - .enable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - assert_eq!( - config.multi_agent_version_from_features(), - Some(MultiAgentVersion::V2) - ); - - Ok(()) -} - #[tokio::test] async fn multi_agent_v2_rejects_agents_max_threads() -> std::io::Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/core/src/session/config_lock.rs b/codex-rs/core/src/session/config_lock.rs index cbb03408bc..f5aadd8e07 100644 --- a/codex-rs/core/src/session/config_lock.rs +++ b/codex-rs/core/src/session/config_lock.rs @@ -21,6 +21,7 @@ use super::SessionConfiguration; pub(crate) async fn validate_config_lock_if_configured( session_configuration: &SessionConfiguration, + multi_agent_version: Option, ) -> anyhow::Result<()> { if session_configuration.session_source.is_non_root_agent() { return Ok(()); @@ -32,7 +33,8 @@ pub(crate) async fn validate_config_lock_if_configured( else { return Ok(()); }; - let actual = session_configuration.to_config_lockfile_toml()?; + let actual = session_configuration + .to_config_lockfile_toml_with_multi_agent_version(multi_agent_version)?; let config = session_configuration.original_config_do_not_use.as_ref(); let options = ConfigLockReplayOptions { allow_codex_version_mismatch: config.config_lock_allow_codex_version_mismatch, @@ -45,13 +47,15 @@ pub(crate) async fn validate_config_lock_if_configured( pub(crate) async fn export_config_lock_if_configured( session_configuration: &SessionConfiguration, conversation_id: ThreadId, + multi_agent_version: Option, ) -> anyhow::Result<()> { let config = session_configuration.original_config_do_not_use.as_ref(); let Some(export_dir) = config.config_lock_export_dir.as_ref() else { return Ok(()); }; - let lock = session_configuration.to_config_lockfile_toml()?; + let lock = session_configuration + .to_config_lockfile_toml_with_multi_agent_version(multi_agent_version)?; let lock = toml::to_string_pretty(&lock).context("failed to serialize config lock")?; let path = export_dir.join(format!("{conversation_id}.config.lock.toml")); @@ -71,15 +75,28 @@ pub(crate) async fn export_config_lock_if_configured( } impl SessionConfiguration { + #[cfg(test)] pub(crate) fn to_config_lockfile_toml(&self) -> anyhow::Result { + self.to_config_lockfile_toml_with_multi_agent_version( + self.original_config_do_not_use + .multi_agent_version_from_features(), + ) + } + + fn to_config_lockfile_toml_with_multi_agent_version( + &self, + multi_agent_version: Option, + ) -> anyhow::Result { Ok(config_lockfile(session_configuration_to_lock_config_toml( self, + multi_agent_version, )?)) } } fn session_configuration_to_lock_config_toml( sc: &SessionConfiguration, + multi_agent_version: Option, ) -> anyhow::Result { let config = sc.original_config_do_not_use.as_ref(); // Start from the resolved layer stack, then patch in values that are only @@ -95,7 +112,7 @@ fn session_configuration_to_lock_config_toml( save_session_resolved_fields(sc, &mut lock_config); } - save_config_resolved_fields(sc, config, &mut lock_config)?; + save_config_resolved_fields(config, multi_agent_version, &mut lock_config)?; drop_lockfile_inputs(&mut lock_config); Ok(lock_config) @@ -125,8 +142,8 @@ fn save_session_resolved_fields(sc: &SessionConfiguration, lock_config: &mut Con /// Persist the resolved representation so replay compares against the behavior /// Codex actually ran with, not only the user-authored TOML inputs. fn save_config_resolved_fields( - sc: &SessionConfiguration, config: &Config, + multi_agent_version: Option, lock_config: &mut ConfigToml, ) -> anyhow::Result<()> { lock_config.web_search = Some(config.web_search_mode.value()); @@ -147,15 +164,15 @@ fn save_config_resolved_fields( .features .get_or_insert_with(FeaturesToml::default); let mut materialized_features = config.features.get().clone(); - materialized_features.set_enabled(Feature::Collab, sc.multi_agent_version.is_some()); + materialized_features.set_enabled(Feature::Collab, multi_agent_version.is_some()); materialized_features.set_enabled( Feature::MultiAgentV2, - sc.multi_agent_version == Some(MultiAgentVersion::V2), + multi_agent_version == Some(MultiAgentVersion::V2), ); features.materialize_resolved_enabled(&materialized_features); let mut multi_agent_v2: MultiAgentV2ConfigToml = resolved_config_to_toml(&config.multi_agent_v2, "features.multi_agent_v2")?; - multi_agent_v2.enabled = Some(sc.multi_agent_version == Some(MultiAgentVersion::V2)); + multi_agent_v2.enabled = Some(multi_agent_version == Some(MultiAgentVersion::V2)); features.multi_agent_v2 = Some(FeatureToml::Config(multi_agent_v2)); features.apps_mcp_path_override = Some(FeatureToml::Config(AppsMcpPathOverrideConfigToml { enabled: Some(config.features.enabled(Feature::AppsMcpPathOverride)), @@ -169,10 +186,10 @@ fn save_config_resolved_fields( let agents = lock_config.agents.get_or_insert_with(Default::default); // Multi-agent v2 owns thread fanout through its feature config. Preserve // the legacy agents.max_threads setting only when v2 is disabled. - agents.max_threads = if sc.multi_agent_version == Some(MultiAgentVersion::V2) { + agents.max_threads = if multi_agent_version == Some(MultiAgentVersion::V2) { None } else { - config.effective_agent_max_threads(sc.multi_agent_version)? + config.effective_agent_max_threads(multi_agent_version)? }; agents.max_depth = Some(config.agent_max_depth); agents.job_max_runtime_seconds = config.agent_job_max_runtime_seconds; @@ -296,8 +313,9 @@ mod tests { (Some(MultiAgentVersion::V1), true, false), (Some(MultiAgentVersion::V2), true, true), ] { - sc.multi_agent_version = multi_agent_version; - let lockfile = sc.to_config_lockfile_toml().expect("lock should serialize"); + let lockfile = sc + .to_config_lockfile_toml_with_multi_agent_version(multi_agent_version) + .expect("lock should serialize"); let lock = &lockfile.config; let entries = lock .features diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index af09c6636a..5f1e69c7eb 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -573,7 +573,6 @@ impl Codex { let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version, model_reasoning_summary: config.model_reasoning_summary, service_tier, developer_instructions: config.developer_instructions.clone(), @@ -629,6 +628,7 @@ impl Codex { thread_store, parent_rollout_thread_trace, attestation_provider, + multi_agent_version, ) .await .map_err(|e| { @@ -2623,7 +2623,7 @@ impl Session { state.session_configuration.collaboration_mode.clone() } - pub(crate) async fn multi_agent_version(&self) -> Option { + pub(crate) fn multi_agent_version(&self) -> Option { self.multi_agent_version.get().copied() } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index d9e8336cf8..ddd35c8b96 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -48,7 +48,6 @@ pub(crate) struct SessionConfiguration { pub(super) provider: ModelProviderInfo, pub(super) collaboration_mode: CollaborationMode, - pub(super) multi_agent_version: Option, pub(super) model_reasoning_summary: Option, pub(super) service_tier: Option, @@ -508,6 +507,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={:?}", @@ -522,6 +522,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 event_persistence_mode = if session_configuration.persist_extended_history { ThreadEventPersistenceMode::Extended @@ -568,7 +570,7 @@ impl Session { text: session_configuration.base_instructions.clone(), }, dynamic_tools: session_configuration.dynamic_tools.clone(), - multi_agent_version: session_configuration.multi_agent_version, + multi_agent_version: initial_multi_agent_version, metadata: ThreadPersistenceMetadata { cwd: Some(config.cwd.to_path_buf()), model_provider: config.model_provider_id.clone(), @@ -886,8 +888,17 @@ impl Session { )) .await; session_configuration.thread_name = thread_name.clone(); - validate_config_lock_if_configured(&session_configuration).await?; - export_config_lock_if_configured(&session_configuration, thread_id).await?; + validate_config_lock_if_configured( + &session_configuration, + initial_multi_agent_version, + ) + .await?; + export_config_lock_if_configured( + &session_configuration, + thread_id, + initial_multi_agent_version, + ) + .await?; let state = SessionState::new(session_configuration.clone()); let managed_network_requirements_configured = config .config_layer_stack @@ -1075,10 +1086,7 @@ impl Session { state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), - multi_agent_version: session_configuration - .multi_agent_version - .map(OnceLock::from) - .unwrap_or_default(), + 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 e78d9faa5c..3560dad664 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -3080,7 +3080,6 @@ async fn set_rate_limits_retains_previous_credits() { let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -3187,7 +3186,6 @@ async fn set_rate_limits_updates_plan_type_when_present() { let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -3719,7 +3717,6 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -4465,7 +4462,6 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -4531,6 +4527,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, + config.multi_agent_version_from_features(), ) .await; @@ -4577,7 +4574,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -4731,7 +4727,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { &session_telemetry, session_configuration.provider.clone(), &session_configuration, - session_configuration.multi_agent_version, + config.multi_agent_version_from_features(), services.user_shell.as_ref(), services.shell_zsh_path.as_ref(), services.main_execve_wrapper_exe.as_ref(), @@ -4755,8 +4751,8 @@ 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: session_configuration - .multi_agent_version + multi_agent_version: config + .multi_agent_version_from_features() .map(OnceLock::from) .unwrap_or_default(), pending_mcp_server_refresh_config: Mutex::new(None), @@ -4820,7 +4816,6 @@ async fn make_session_with_config_and_rx( let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -4887,6 +4882,7 @@ async fn make_session_with_config_and_rx( )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, + config.multi_agent_version_from_features(), ) .await?; @@ -4926,7 +4922,6 @@ async fn make_session_with_history_source_and_agent_control_and_rx( let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -5000,6 +4995,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( )), codex_rollout_trace::ThreadTraceContext::disabled(), /*attestation_provider*/ None, + config.multi_agent_version_from_features(), ) .await?; @@ -6434,7 +6430,6 @@ where let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, - multi_agent_version: config.multi_agent_version_from_features(), model_reasoning_summary: config.model_reasoning_summary, developer_instructions: config.developer_instructions.clone(), user_instructions: config.user_instructions.clone(), @@ -6588,7 +6583,7 @@ where &session_telemetry, session_configuration.provider.clone(), &session_configuration, - session_configuration.multi_agent_version, + config.multi_agent_version_from_features(), services.user_shell.as_ref(), services.shell_zsh_path.as_ref(), services.main_execve_wrapper_exe.as_ref(), @@ -6612,8 +6607,8 @@ where state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), - multi_agent_version: session_configuration - .multi_agent_version + multi_agent_version: config + .multi_agent_version_from_features() .map(OnceLock::from) .unwrap_or_default(), pending_mcp_server_refresh_config: 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 06d65c495a..4dec5e1eed 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -730,7 +730,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { }], } ); - assert_eq!(codex.session.multi_agent_version().await, None); + assert_eq!(codex.session.multi_agent_version(), None); drop(codex); } diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index f6df06fac1..c02d524e88 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -732,7 +732,7 @@ impl Session { &self.services.session_telemetry, session_configuration.provider.clone(), &session_configuration, - self.multi_agent_version().await, + self.multi_agent_version(), self.services.user_shell.as_ref(), self.services.shell_zsh_path.as_ref(), self.services.main_execve_wrapper_exe.as_ref(), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 21787d828c..12f23d7bc3 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -657,10 +657,9 @@ 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().await.ok_or_else(|| { - CodexErr::InvalidRequest("multi-agent version is unresolved".to_string()) - })?; + let inherited_multi_agent_version = fork_source.multi_agent_version().ok_or_else(|| { + CodexErr::InvalidRequest("multi-agent version is unresolved".to_string()) + })?; options.initial_history = fork_history_from_snapshot( ForkSnapshot::Interrupted, history, @@ -1243,7 +1242,7 @@ impl ThreadManagerState { None => None, }; let live_multi_agent_version = match source_thread.as_ref() { - Some(source_thread) => source_thread.multi_agent_version().await, + Some(source_thread) => source_thread.multi_agent_version(), None => None, }; let multi_agent_version = live_multi_agent_version @@ -1367,14 +1366,18 @@ 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 - .resolve_multi_agent_version( - &config, - &initial_history, - forked_from_thread_id, - inherited_multi_agent_version, - ) - .await; + let multi_agent_version = match inherited_multi_agent_version { + Some(multi_agent_version) => Some(multi_agent_version), + None => { + self.resolve_multi_agent_version( + &config, + &initial_history, + forked_from_thread_id, + /*inherited_multi_agent_version*/ None, + ) + .await + } + }; let CodexSpawnOk { codex, thread_id, .. } = Codex::spawn(CodexSpawnArgs { diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index d40d2370e6..742192bd7f 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -804,7 +804,7 @@ async fn resume_and_fork_preserve_stored_multi_agent_version() { .await .expect("start source thread"); assert_eq!( - source.thread.multi_agent_version().await, + source.thread.multi_agent_version(), Some(MultiAgentVersion::V1) ); source.thread.ensure_rollout_materialized().await; @@ -839,7 +839,7 @@ async fn resume_and_fork_preserve_stored_multi_agent_version() { .await .expect("resume source thread"); assert_eq!( - resumed.thread.multi_agent_version().await, + resumed.thread.multi_agent_version(), Some(MultiAgentVersion::V1) ); resumed @@ -861,7 +861,7 @@ async fn resume_and_fork_preserve_stored_multi_agent_version() { .await .expect("fork source thread"); assert_eq!( - forked.thread.multi_agent_version().await, + forked.thread.multi_agent_version(), Some(MultiAgentVersion::V1) ); forked @@ -929,7 +929,7 @@ async fn legacy_resume_continues_when_multi_agent_version_seed_fails() { .await .expect("legacy resume should continue after seed failure"); assert_eq!( - resumed.thread.multi_agent_version().await, + resumed.thread.multi_agent_version(), Some(MultiAgentVersion::V2) ); assert_eq!( 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 045e448d75..7a75ae1a52 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -1363,7 +1363,6 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() { let multi_agent_version = root .thread .multi_agent_version() - .await .expect("root thread should have a multi-agent version"); let child_path = AgentPath::try_from("/root/worker").expect("agent path"); @@ -1445,7 +1444,6 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() { let multi_agent_version = root .thread .multi_agent_version() - .await .expect("root thread should have a multi-agent version"); let child_path = AgentPath::try_from("/root/worker").expect("agent path"); @@ -1625,7 +1623,6 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() { let multi_agent_version = root .thread .multi_agent_version() - .await .expect("root thread should have a multi-agent version"); let researcher_path = AgentPath::from_string("/root/researcher".to_string()).expect("path"); diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 329c1406e0..53dd623e9c 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2795,7 +2795,7 @@ pub struct SessionMeta { pub dynamic_tools: Option>, #[serde(skip_serializing_if = "Option::is_none")] pub memory_mode: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] + #[serde(skip_serializing_if = "Option::is_none")] pub multi_agent_version: Option, } diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 4e7b76bbc6..0b3830e141 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -48,105 +48,6 @@ mod tests { use crate::ListTurnsParams; use crate::SortDirection; use crate::StoredTurnItemsView; - use crate::ThreadEventPersistenceMode; - use crate::ThreadPersistenceMetadata; - use codex_protocol::models::BaseInstructions; - use codex_protocol::protocol::SessionSource; - use pretty_assertions::assert_eq; - - #[tokio::test] - async fn create_thread_records_multi_agent_version() { - let store = InMemoryThreadStore::default(); - let thread_id = ThreadId::new(); - store - .create_thread(CreateThreadParams { - thread_id, - forked_from_id: None, - source: SessionSource::Exec, - thread_source: None, - base_instructions: BaseInstructions::default(), - dynamic_tools: Vec::new(), - multi_agent_version: Some(MultiAgentVersion::V2), - metadata: ThreadPersistenceMetadata { - cwd: None, - model_provider: "test-provider".to_string(), - memory_mode: ThreadMemoryMode::Enabled, - }, - event_persistence_mode: ThreadEventPersistenceMode::Limited, - }) - .await - .expect("create thread should succeed"); - - let history = store - .load_history(LoadThreadHistoryParams { - thread_id, - include_archived: true, - }) - .await - .expect("created history should load"); - let versions = history - .items - .into_iter() - .filter_map(|item| match item { - RolloutItem::SessionMeta(meta_line) => meta_line.meta.multi_agent_version, - _ => None, - }) - .collect::>(); - assert_eq!(versions, vec![MultiAgentVersion::V2]); - } - - #[tokio::test] - async fn concurrent_multi_agent_version_seeds_return_the_first_winner() { - let store = InMemoryThreadStore::default(); - let thread_id = ThreadId::new(); - store.state.lock().await.histories.insert( - thread_id, - vec![RolloutItem::SessionMeta(SessionMetaLine { - meta: SessionMeta { - id: thread_id, - ..SessionMeta::default() - }, - git: None, - })], - ); - - let (left, right) = tokio::join!( - store.set_multi_agent_version_if_unset(SetMultiAgentVersionIfUnsetParams { - thread_id, - multi_agent_version: MultiAgentVersion::V1, - include_archived: true, - }), - store.set_multi_agent_version_if_unset(SetMultiAgentVersionIfUnsetParams { - thread_id, - multi_agent_version: MultiAgentVersion::V2, - include_archived: true, - }), - ); - let left = left.expect("first seed should succeed"); - let right = right.expect("second seed should return the stored winner"); - assert_eq!(left, right); - assert!(matches!( - left, - MultiAgentVersion::V1 | MultiAgentVersion::V2 - )); - - let history = store - .load_history(LoadThreadHistoryParams { - thread_id, - include_archived: true, - }) - .await - .expect("seeded history should load"); - let versions = history - .items - .into_iter() - .filter_map(|item| match item { - RolloutItem::SessionMeta(meta_line) => meta_line.meta.multi_agent_version, - _ => None, - }) - .collect::>(); - assert_eq!(versions, vec![left]); - } #[tokio::test] async fn default_turn_pagination_methods_return_unsupported() { diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 4e3d37db64..fda42f308d 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -21,7 +21,6 @@ use std::collections::hash_map::Entry; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; -use tokio::sync::Semaphore; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; @@ -60,7 +59,7 @@ use codex_protocol::protocol::MultiAgentVersion; pub struct LocalThreadStore { pub(super) config: LocalThreadStoreConfig, live_recorders: Arc>>, - multi_agent_version_seed_lock: Arc, + multi_agent_version_seed_lock: Arc>, state_db: Option, } @@ -100,7 +99,7 @@ impl LocalThreadStore { Self { config, live_recorders: Arc::new(Mutex::new(HashMap::new())), - multi_agent_version_seed_lock: Arc::new(Semaphore::new(/*permits*/ 1)), + multi_agent_version_seed_lock: Arc::new(Mutex::new(())), state_db, } } diff --git a/codex-rs/thread-store/src/local/set_multi_agent_version.rs b/codex-rs/thread-store/src/local/set_multi_agent_version.rs index 2c5cf42f0a..d077d073c3 100644 --- a/codex-rs/thread-store/src/local/set_multi_agent_version.rs +++ b/codex-rs/thread-store/src/local/set_multi_agent_version.rs @@ -16,13 +16,7 @@ pub(super) async fn set_multi_agent_version_if_unset( store: &LocalThreadStore, params: SetMultiAgentVersionIfUnsetParams, ) -> ThreadStoreResult { - let _permit = store - .multi_agent_version_seed_lock - .acquire() - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to lock thread multi-agent version seed: {err}"), - })?; + let _guard = store.multi_agent_version_seed_lock.lock().await; let has_live_writer = live_writer::rollout_path(store, params.thread_id) .await .is_ok(); @@ -63,14 +57,6 @@ pub(super) async fn set_multi_agent_version_if_unset( .map_err(|err| ThreadStoreError::Internal { message: format!("failed to set thread multi-agent version: {err}"), })?; - if session_meta.meta.id != params.thread_id { - return Err(ThreadStoreError::Internal { - message: format!( - "failed to set thread multi-agent version: rollout session metadata id mismatch: expected {}, found {}", - params.thread_id, session_meta.meta.id - ), - }); - } session_meta.git = None; session_meta.meta.multi_agent_version = Some(params.multi_agent_version); let item = RolloutItem::SessionMeta(session_meta);