codex: move multi-agent regression coverage to integration suite

This commit is contained in:
Ahmed Ibrahim
2026-06-01 13:13:17 -07:00
parent 19d9486974
commit ab6593cf38
6 changed files with 134 additions and 276 deletions
-48
View File
@@ -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;
@@ -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);
}
-183
View File
@@ -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::<InMemoryThreadStore>()
.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");
@@ -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| {
+134
View File
@@ -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<RolloutItem> {
}
items
}
fn latest_multi_agent_version(
path: &std::path::Path,
thread_id: ThreadId,
) -> Option<MultiAgentVersion> {
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::<RolloutLine>(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()
}
-2
View File
@@ -104,7 +104,6 @@ fn stores_guard() -> MutexGuard<'static, HashMap<String, Arc<InMemoryThreadStore
pub struct InMemoryThreadStoreCalls {
pub create_thread: usize,
pub resume_thread: usize,
pub set_multi_agent_version_if_unset: usize,
pub append_items: usize,
pub persist_thread: usize,
pub flush_thread: usize,
@@ -214,7 +213,6 @@ impl ThreadStore for InMemoryThreadStore {
params: SetMultiAgentVersionIfUnsetParams,
) -> ThreadStoreResult<MultiAgentVersion> {
let mut state = self.state.lock().await;
state.calls.set_multi_agent_version_if_unset += 1;
let history =
state
.histories