From ab6593cf38b031f34b77a8b15576de6e179e7024 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Sat, 30 May 2026 22:25:32 -0700 Subject: [PATCH] codex: move multi-agent regression coverage to integration suite --- codex-rs/core/src/agent/control_tests.rs | 48 ----- .../core/src/session/tests/guardian_tests.rs | 2 - codex-rs/core/src/thread_manager_tests.rs | 183 ------------------ codex-rs/core/src/tools/spec_plan_tests.rs | 41 ---- codex-rs/core/tests/suite/fork_thread.rs | 134 +++++++++++++ codex-rs/thread-store/src/in_memory.rs | 2 - 6 files changed, 134 insertions(+), 276 deletions(-) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 2d00fc2845..fbd84d4ddc 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -21,7 +21,6 @@ use codex_protocol::protocol::CompactedItem; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InterAgentCommunication; -use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::TurnAbortReason; @@ -567,53 +566,6 @@ async fn spawn_agent_creates_thread_and_sends_prompt() { assert_eq!(captured, Some(expected)); } -#[tokio::test] -async fn spawn_agent_inherits_parent_multi_agent_version_over_child_feature_flags() { - let harness = AgentControlHarness::new().await; - let mut parent_config = harness.config.clone(); - parent_config - .features - .enable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - let parent = harness - .manager - .start_thread(parent_config) - .await - .expect("start parent thread"); - let mut child_config = harness.config.clone(); - child_config - .features - .disable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - child_config - .features - .disable(Feature::Collab) - .expect("test config should allow feature update"); - - let child_thread_id = harness - .control - .spawn_agent( - child_config, - text_input("spawned"), - Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id: parent.thread_id, - depth: 1, - agent_path: None, - agent_nickname: None, - agent_role: None, - })), - ) - .await - .expect("spawn_agent should succeed"); - let child = harness - .manager - .get_thread(child_thread_id) - .await - .expect("child thread should be registered"); - - assert_eq!(child.multi_agent_version(), Some(MultiAgentVersion::V2)); -} - #[tokio::test] async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { let harness = AgentControlHarness::new().await; diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 4dec5e1eed..2c64144a3e 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -730,7 +730,5 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { }], } ); - assert_eq!(codex.session.multi_agent_version(), None); - drop(codex); } diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 1c56227d90..e768910d51 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -17,10 +17,7 @@ use codex_protocol::openai_models::ModelsResponse; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InternalSessionSource; -use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::ResumedHistory; -use codex_protocol::protocol::SessionMeta; -use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TurnStartedEvent; @@ -768,186 +765,6 @@ async fn resume_stopped_thread_from_rollout_spawns_new_thread() { .expect("shutdown resumed thread"); } -#[tokio::test] -async fn legacy_resume_seeds_multi_agent_version_for_later_forks() { - let temp_dir = tempdir().expect("tempdir"); - let mut config = test_config().await; - config.codex_home = temp_dir.path().join("codex-home").abs(); - config.cwd = config.codex_home.abs(); - config - .features - .disable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - config - .features - .disable(Feature::Collab) - .expect("test config should allow feature update"); - std::fs::create_dir_all(&config.codex_home).expect("create codex home"); - - let auth_manager = - AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); - let manager = ThreadManager::new( - &config, - auth_manager.clone(), - SessionSource::Exec, - Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), - empty_extension_registry(), - /*analytics_events_client*/ None, - thread_store_from_config(&config, /*state_db*/ None), - /*state_db*/ None, - TEST_INSTALLATION_ID.to_string(), - /*attestation_provider*/ None, - ); - - let source = manager - .start_thread(config.clone()) - .await - .expect("start source thread"); - assert_eq!(source.thread.multi_agent_version(), None); - source.thread.ensure_rollout_materialized().await; - source - .thread - .flush_rollout() - .await - .expect("flush source rollout"); - let rollout_path = source - .thread - .rollout_path() - .expect("source rollout path should exist"); - source - .thread - .shutdown_and_wait() - .await - .expect("shutdown source thread"); - let _ = manager.remove_thread(&source.thread_id).await; - - let mut v1_config = config; - v1_config - .features - .enable(Feature::Collab) - .expect("test config should allow feature update"); - let resumed = manager - .resume_thread_from_rollout( - v1_config.clone(), - rollout_path.clone(), - auth_manager, - /*parent_trace*/ None, - ) - .await - .expect("resume legacy source thread"); - assert_eq!( - resumed.thread.multi_agent_version(), - Some(MultiAgentVersion::V1) - ); - resumed - .thread - .shutdown_and_wait() - .await - .expect("shutdown resumed thread"); - let _ = manager.remove_thread(&resumed.thread_id).await; - - let mut v2_config = v1_config; - v2_config - .features - .enable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - let forked = manager - .fork_thread( - ForkSnapshot::Interrupted, - v2_config, - rollout_path, - /*thread_source*/ None, - /*persist_extended_history*/ false, - /*parent_trace*/ None, - ) - .await - .expect("fork source thread"); - assert_eq!( - forked.thread.multi_agent_version(), - Some(MultiAgentVersion::V1) - ); - forked - .thread - .shutdown_and_wait() - .await - .expect("shutdown forked thread"); -} - -#[tokio::test] -async fn legacy_resume_continues_when_multi_agent_version_seed_fails() { - let temp_dir = tempdir().expect("tempdir"); - let mut config = test_config().await; - config.codex_home = temp_dir.path().join("codex-home").abs(); - config.cwd = config.codex_home.abs(); - config.experimental_thread_store = ThreadStoreConfig::InMemory { - id: format!("thread-manager-{}", uuid::Uuid::new_v4()), - }; - config - .features - .enable(Feature::MultiAgentV2) - .expect("test config should allow feature update"); - std::fs::create_dir_all(&config.codex_home).expect("create codex home"); - - let auth_manager = - AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); - let thread_store = thread_store_from_config(&config, /*state_db*/ None); - let in_memory_store = thread_store - .as_any() - .downcast_ref::() - .expect("configured in-memory store"); - let manager = ThreadManager::new( - &config, - auth_manager.clone(), - SessionSource::Exec, - Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), - empty_extension_registry(), - /*analytics_events_client*/ None, - Arc::clone(&thread_store), - /*state_db*/ None, - TEST_INSTALLATION_ID.to_string(), - /*attestation_provider*/ None, - ); - let thread_id = ThreadId::new(); - - let resumed = manager - .resume_thread_with_history( - config, - InitialHistory::Resumed(ResumedHistory { - conversation_id: thread_id, - history: vec![RolloutItem::SessionMeta(SessionMetaLine { - meta: SessionMeta { - id: thread_id, - cwd: temp_dir.path().to_path_buf(), - ..SessionMeta::default() - }, - git: None, - })], - rollout_path: None, - }), - auth_manager, - /*persist_extended_history*/ false, - /*parent_trace*/ None, - ) - .await - .expect("legacy resume should continue after seed failure"); - assert_eq!( - resumed.thread.multi_agent_version(), - Some(MultiAgentVersion::V2) - ); - assert_eq!( - in_memory_store - .calls() - .await - .set_multi_agent_version_if_unset, - 1 - ); - resumed - .thread - .shutdown_and_wait() - .await - .expect("shutdown resumed thread"); -} - #[tokio::test] async fn resume_stopped_thread_from_rollout_preserves_thread_source() { let temp_dir = tempdir().expect("tempdir"); diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index e0aa46c69c..df84ce7966 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -15,7 +15,6 @@ use codex_protocol::openai_models::ConfigShellToolType; use codex_protocol::openai_models::InputModality; use codex_protocol::openai_models::ToolMode; use codex_protocol::openai_models::WebSearchToolType; -use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_tools::DiscoverablePluginInfo; @@ -530,14 +529,6 @@ async fn host_context_gates_goal_and_agent_job_tools() { normal_agent_job.assert_visible_contains(&["spawn_agents_on_csv"]); normal_agent_job.assert_visible_lacks(&["report_agent_job_result"]); - let agent_job_without_multi_agent = probe(|turn| { - set_feature(turn, Feature::SpawnCsv, /*enabled*/ true); - turn.multi_agent_version = None; - }) - .await; - agent_job_without_multi_agent.assert_visible_lacks(&["spawn_agents_on_csv"]); - agent_job_without_multi_agent.assert_visible_lacks(&[MULTI_AGENT_V1_NAMESPACE, "spawn_agent"]); - let worker_agent_job = probe(|turn| { set_feature(turn, Feature::SpawnCsv, /*enabled*/ true); turn.session_source = @@ -874,38 +865,6 @@ async fn multi_agent_feature_selects_one_agent_tool_family() { ); } -#[tokio::test] -async fn multi_agent_tool_family_uses_locked_version_instead_of_features() { - let v1 = probe(|turn| { - set_feature(turn, Feature::MultiAgentV2, /*enabled*/ true); - turn.multi_agent_version = Some(MultiAgentVersion::V1); - }) - .await; - v1.assert_visible_contains(&[MULTI_AGENT_V1_NAMESPACE]); - v1.assert_visible_lacks(&["spawn_agent", "send_message", "assign_task", "list_agents"]); - - let v2 = probe(|turn| { - set_feature(turn, Feature::Collab, /*enabled*/ false); - turn.multi_agent_version = Some(MultiAgentVersion::V2); - }) - .await; - v2.assert_visible_contains(&["spawn_agent", "send_message", "assign_task", "list_agents"]); - v2.assert_visible_lacks(&[MULTI_AGENT_V1_NAMESPACE]); - - let disabled = probe(|turn| { - set_feature(turn, Feature::Collab, /*enabled*/ true); - turn.multi_agent_version = None; - }) - .await; - disabled.assert_visible_lacks(&[ - MULTI_AGENT_V1_NAMESPACE, - "spawn_agent", - "send_message", - "assign_task", - "list_agents", - ]); -} - #[tokio::test] async fn tool_mode_selector_overrides_feature_flags() { let direct = probe(|turn| { diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index 544c530aa0..2d1532d678 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -1,9 +1,12 @@ use codex_core::ForkSnapshot; use codex_core::NewThread; use codex_core::parse_turn_item; +use codex_features::Feature; +use codex_protocol::ThreadId; use codex_protocol::items::TurnItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InitialHistory; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::RolloutItem; @@ -15,12 +18,121 @@ use core_test_support::responses::sse; use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use pretty_assertions::assert_eq; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_resume_seeds_multi_agent_version_for_later_forks() { + skip_if_no_network!(); + + let server = MockServer::start().await; + let mut initial_builder = test_codex().with_config(|config| { + config + .features + .disable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + config + .features + .disable(Feature::Collab) + .expect("test config should allow feature update"); + }); + let initial = initial_builder + .build(&server) + .await + .expect("create legacy source thread"); + assert_eq!(initial.codex.multi_agent_version(), None); + initial.codex.ensure_rollout_materialized().await; + initial + .codex + .flush_rollout() + .await + .expect("flush source rollout"); + let source_thread_id = initial.session_configured.thread_id; + let rollout_path = initial.codex.rollout_path().expect("source rollout path"); + initial + .codex + .shutdown_and_wait() + .await + .expect("shutdown source thread"); + + let mut resume_builder = test_codex().with_config(|config| { + config + .features + .disable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + config + .features + .enable(Feature::Collab) + .expect("test config should allow feature update"); + }); + let resumed = resume_builder + .resume(&server, initial.home.clone(), rollout_path.clone()) + .await + .expect("resume legacy source thread"); + assert_eq!( + resumed.codex.multi_agent_version(), + Some(MultiAgentVersion::V1) + ); + resumed + .codex + .flush_rollout() + .await + .expect("flush resumed rollout"); + assert_eq!( + latest_multi_agent_version(&rollout_path, source_thread_id), + Some(MultiAgentVersion::V1) + ); + resumed + .codex + .shutdown_and_wait() + .await + .expect("shutdown resumed thread"); + + let mut v2_config = resumed.config.clone(); + v2_config + .features + .enable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + let forked = resumed + .thread_manager + .fork_thread( + ForkSnapshot::Interrupted, + v2_config, + rollout_path, + /*thread_source*/ None, + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ) + .await + .expect("fork seeded source thread"); + assert_eq!( + forked.thread.multi_agent_version(), + Some(MultiAgentVersion::V1) + ); + forked.thread.ensure_rollout_materialized().await; + forked + .thread + .flush_rollout() + .await + .expect("flush forked rollout"); + assert_eq!( + latest_multi_agent_version( + &forked.thread.rollout_path().expect("forked rollout path"), + forked.thread_id, + ), + Some(MultiAgentVersion::V1) + ); + forked + .thread + .shutdown_and_wait() + .await + .expect("shutdown forked thread"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn fork_thread_twice_drops_to_first_message() { skip_if_no_network!(); @@ -251,3 +363,25 @@ fn read_rollout_items(path: &std::path::Path) -> Vec { } items } + +fn latest_multi_agent_version( + path: &std::path::Path, + thread_id: ThreadId, +) -> Option { + std::fs::read_to_string(path) + .unwrap_or_else(|err| panic!("failed to read rollout file {}: {err}", path.display())) + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| { + serde_json::from_str::(line) + .unwrap_or_else(|err| panic!("failed to parse rollout line `{line}`: {err}")) + }) + .filter_map(|line| match line.item { + RolloutItem::SessionMeta(meta_line) if meta_line.meta.id == thread_id => { + Some(meta_line.meta.multi_agent_version) + } + _ => None, + }) + .last() + .flatten() +} diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 0b3830e141..355d549030 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -104,7 +104,6 @@ fn stores_guard() -> MutexGuard<'static, HashMap ThreadStoreResult { let mut state = self.state.lock().await; - state.calls.set_multi_agent_version_if_unset += 1; let history = state .histories