openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

codex: trim multi-agent version plumbing

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-06-01 13:13:17 -07:00
parent 3f82d5bd99
commit 6b8fdbdf70
18 changed files with 91 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/common/rollout.rs
+2 -3
View File
@@ -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,
codex-rs/core/src/agent/control.rs
+4 -5
View File
@@ -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,
};
codex-rs/core/src/agent/control_tests.rs
+1 -10
View File
@@ -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),
codex-rs/core/src/codex_thread.rs
+2 -2
View File
@@ -503,8 +503,8 @@ impl CodexThread {
self.codex.session.get_config().await
}
pub async fn multi_agent_version(&self) -> Option<MultiAgentVersion> {
self.codex.session.multi_agent_version().await
pub fn multi_agent_version(&self) -> Option<MultiAgentVersion> {
self.codex.session.multi_agent_version()
}
pub(crate) fn set_multi_agent_version_if_unset(
codex-rs/core/src/config/config_tests.rs
-34
View File
@@ -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()?;
codex-rs/core/src/session/config_lock.rs
+29 -11
View File
@@ -21,6 +21,7 @@ use super::SessionConfiguration;
pub(crate) async fn validate_config_lock_if_configured(
session_configuration: &SessionConfiguration,
multi_agent_version: Option<MultiAgentVersion>,
) -> 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<MultiAgentVersion>,
) -> 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<ConfigLockfileToml> {
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<MultiAgentVersion>,
) -> anyhow::Result<ConfigLockfileToml> {
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<MultiAgentVersion>,
) -> anyhow::Result<ConfigToml> {
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<MultiAgentVersion>,
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
codex-rs/core/src/session/mod.rs
+2 -2
View File
@@ -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<MultiAgentVersion> {
pub(crate) fn multi_agent_version(&self) -> Option<MultiAgentVersion> {
self.multi_agent_version.get().copied()
}
codex-rs/core/src/session/session.rs
+16 -8
View File
@@ -48,7 +48,6 @@ pub(crate) struct SessionConfiguration {
pub(super) provider: ModelProviderInfo,
pub(super) collaboration_mode: CollaborationMode,
pub(super) multi_agent_version: Option<MultiAgentVersion>,
pub(super) model_reasoning_summary: Option<ReasoningSummaryConfig>,
pub(super) service_tier: Option<String>,
@@ -508,6 +507,7 @@ impl Session {
thread_store: Arc<dyn ThreadStore>,
parent_rollout_thread_trace: ThreadTraceContext,
attestation_provider: Option<Arc<dyn AttestationProvider>>,
multi_agent_version: Option<MultiAgentVersion>,
) -> anyhow::Result<Arc<Self>> {
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),
codex-rs/core/src/session/tests.rs
+9 -14
View File
@@ -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),
codex-rs/core/src/session/tests/guardian_tests.rs
+1 -1
View File
@@ -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);
}
codex-rs/core/src/session/turn_context.rs
+1 -1
View File
@@ -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(),
codex-rs/core/src/thread_manager.rs
+16 -13
View File
@@ -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 {
codex-rs/core/src/thread_manager_tests.rs
+4 -4
View File
@@ -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!(
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
-3
View File
@@ -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");
codex-rs/protocol/src/protocol.rs
+1 -1
View File
@@ -2795,7 +2795,7 @@ pub struct SessionMeta {
pub dynamic_tools: Option<Vec<DynamicToolSpec>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub memory_mode: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[serde(skip_serializing_if = "Option::is_none")]
pub multi_agent_version: Option<MultiAgentVersion>,
}
codex-rs/thread-store/src/in_memory.rs
-99
View File
@@ -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::<Vec<_>>();
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::<Vec<_>>();
assert_eq!(versions, vec![left]);
}
#[tokio::test]
async fn default_turn_pagination_methods_return_unsupported() {
codex-rs/thread-store/src/local/mod.rs
+2 -3
View File
@@ -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<Mutex<HashMap<ThreadId, RolloutRecorder>>>,
multi_agent_version_seed_lock: Arc<Semaphore>,
multi_agent_version_seed_lock: Arc<Mutex<()>>,
state_db: Option<StateDbHandle>,
}
@@ -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,
}
}
codex-rs/thread-store/src/local/set_multi_agent_version.rs
+1 -15
View File
@@ -16,13 +16,7 @@ pub(super) async fn set_multi_agent_version_if_unset(
store: &LocalThreadStore,
params: SetMultiAgentVersionIfUnsetParams,
) -> ThreadStoreResult<MultiAgentVersion> {
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);