mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
codex: pin multi-agent system per thread
This commit is contained in:
@@ -6,6 +6,7 @@ use crate::agent::role::resolve_role_config;
|
||||
use crate::agent::status::is_final;
|
||||
use crate::codex_thread::ThreadConfigSnapshot;
|
||||
use crate::session::emit_subagent_session_started;
|
||||
use crate::session::turn_context::resolve_multi_agent_version;
|
||||
use crate::session_prefix::format_subagent_context_line;
|
||||
use crate::session_prefix::format_subagent_notification_message;
|
||||
use crate::shell_snapshot::ShellSnapshot;
|
||||
@@ -21,6 +22,7 @@ use codex_protocol::error::Result as CodexResult;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::MessagePhase;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::protocol::InitialHistory;
|
||||
use codex_protocol::protocol::InterAgentCommunication;
|
||||
use codex_protocol::protocol::Op;
|
||||
@@ -55,6 +57,8 @@ pub(crate) struct SpawnAgentOptions {
|
||||
pub(crate) fork_parent_spawn_call_id: Option<String>,
|
||||
pub(crate) fork_mode: Option<SpawnAgentForkMode>,
|
||||
pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
|
||||
/// Parent turn's resolved selector.
|
||||
pub(crate) multi_agent_version: Option<MultiAgentVersion>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
@@ -246,6 +250,16 @@ impl AgentControl {
|
||||
}
|
||||
other => (other, AgentMetadata::default()),
|
||||
};
|
||||
let multi_agent_version =
|
||||
session_source
|
||||
.as_ref()
|
||||
.map_or(options.multi_agent_version, |session_source| {
|
||||
resolve_multi_agent_version(
|
||||
&config,
|
||||
session_source,
|
||||
options.multi_agent_version,
|
||||
)
|
||||
});
|
||||
let notification_source = session_source.clone();
|
||||
|
||||
// The same `AgentControl` is sent to spawn the thread.
|
||||
@@ -267,6 +281,7 @@ impl AgentControl {
|
||||
config.clone(),
|
||||
self.clone(),
|
||||
session_source,
|
||||
multi_agent_version,
|
||||
forked_from_thread_id,
|
||||
/*thread_source*/ Some(ThreadSource::Subagent),
|
||||
/*persist_extended_history*/ false,
|
||||
@@ -339,7 +354,7 @@ impl AgentControl {
|
||||
|
||||
self.send_input(new_thread.thread_id, initial_operation)
|
||||
.await?;
|
||||
if !new_thread.thread.enabled(Feature::MultiAgentV2) {
|
||||
if multi_agent_version != Some(MultiAgentVersion::V2) {
|
||||
let child_reference = agent_metadata
|
||||
.agent_path
|
||||
.as_ref()
|
||||
@@ -369,6 +384,8 @@ impl AgentControl {
|
||||
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
|
||||
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
|
||||
) -> CodexResult<crate::thread_manager::NewThread> {
|
||||
let multi_agent_version =
|
||||
resolve_multi_agent_version(&config, &session_source, options.multi_agent_version);
|
||||
if options.fork_parent_spawn_call_id.is_none() {
|
||||
return Err(CodexErr::Fatal(
|
||||
"spawn_agent fork requires a parent spawn call id".to_string(),
|
||||
@@ -418,7 +435,7 @@ impl AgentControl {
|
||||
}
|
||||
let multi_agent_v2_usage_hint_texts_to_filter: Vec<String> =
|
||||
if let Some(parent_thread) = parent_thread.as_ref() {
|
||||
if parent_thread.enabled(Feature::MultiAgentV2) {
|
||||
if multi_agent_version == Some(MultiAgentVersion::V2) {
|
||||
let parent_config = parent_thread.codex.session.get_config().await;
|
||||
[
|
||||
parent_config
|
||||
@@ -436,7 +453,7 @@ impl AgentControl {
|
||||
} else {
|
||||
Vec::new()
|
||||
}
|
||||
} else if config.features.enabled(Feature::MultiAgentV2) {
|
||||
} else if multi_agent_version == Some(MultiAgentVersion::V2) {
|
||||
[
|
||||
config.multi_agent_v2.root_agent_usage_hint_text.clone(),
|
||||
config.multi_agent_v2.subagent_usage_hint_text.clone(),
|
||||
@@ -472,7 +489,7 @@ impl AgentControl {
|
||||
}
|
||||
}
|
||||
if preserve_reference_context_item
|
||||
&& config.features.enabled(Feature::MultiAgentV2)
|
||||
&& multi_agent_version == Some(MultiAgentVersion::V2)
|
||||
&& let Some(subagent_usage_hint_text) =
|
||||
config.multi_agent_v2.subagent_usage_hint_text.clone()
|
||||
&& let Some(subagent_usage_hint_message) =
|
||||
@@ -489,6 +506,7 @@ impl AgentControl {
|
||||
InitialHistory::Forked(forked_rollout_items),
|
||||
self.clone(),
|
||||
session_source,
|
||||
multi_agent_version,
|
||||
/*thread_source*/ Some(ThreadSource::Subagent),
|
||||
/*forked_from_thread_id*/ Some(parent_thread_id),
|
||||
/*persist_extended_history*/ false,
|
||||
@@ -544,11 +562,18 @@ impl AgentControl {
|
||||
let child_resumed = if state.get_thread(child_thread_id).await.is_ok() {
|
||||
true
|
||||
} else {
|
||||
let agent_path = state_db_ctx
|
||||
.get_thread(child_thread_id)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.and_then(|metadata| metadata.agent_path)
|
||||
.and_then(|agent_path| AgentPath::try_from(agent_path).ok());
|
||||
let child_session_source =
|
||||
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id,
|
||||
depth: child_depth,
|
||||
agent_path: None,
|
||||
agent_path,
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
});
|
||||
@@ -581,14 +606,26 @@ impl AgentControl {
|
||||
thread_id: ThreadId,
|
||||
session_source: SessionSource,
|
||||
) -> CodexResult<ThreadId> {
|
||||
let state = self.upgrade()?;
|
||||
let stored_thread = state
|
||||
.read_stored_thread(ReadThreadParams {
|
||||
thread_id,
|
||||
include_archived: true,
|
||||
include_history: true,
|
||||
})
|
||||
.await?;
|
||||
let multi_agent_version = resolve_multi_agent_version(
|
||||
&config,
|
||||
&session_source,
|
||||
stored_thread.multi_agent_version,
|
||||
);
|
||||
if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = &session_source
|
||||
&& *depth >= config.agent_max_depth
|
||||
&& !config.features.enabled(Feature::MultiAgentV2)
|
||||
&& multi_agent_version != Some(MultiAgentVersion::V2)
|
||||
{
|
||||
let _ = config.features.disable(Feature::SpawnCsv);
|
||||
let _ = config.features.disable(Feature::Collab);
|
||||
}
|
||||
let state = self.upgrade()?;
|
||||
let state_db_ctx = state.state_db();
|
||||
let mut reservation = self.state.reserve_spawn_slot(config.agent_max_threads)?;
|
||||
let (session_source, agent_metadata) = match session_source {
|
||||
@@ -627,13 +664,6 @@ impl AgentControl {
|
||||
let inherited_exec_policy = self
|
||||
.inherited_exec_policy_for_source(&state, Some(&session_source), &config)
|
||||
.await;
|
||||
let stored_thread = state
|
||||
.read_stored_thread(ReadThreadParams {
|
||||
thread_id,
|
||||
include_archived: true,
|
||||
include_history: true,
|
||||
})
|
||||
.await?;
|
||||
let history = stored_thread
|
||||
.history
|
||||
.ok_or_else(|| CodexErr::ThreadNotFound(thread_id))?
|
||||
@@ -659,7 +689,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.enabled(Feature::MultiAgentV2) {
|
||||
if multi_agent_version != Some(MultiAgentVersion::V2) {
|
||||
let child_reference = agent_metadata
|
||||
.agent_path
|
||||
.as_ref()
|
||||
@@ -1047,7 +1077,9 @@ impl AgentControl {
|
||||
if child_agent_path.is_some()
|
||||
&& child_thread
|
||||
.as_ref()
|
||||
.map(|thread| thread.enabled(Feature::MultiAgentV2))
|
||||
.map(|thread| {
|
||||
thread.codex.session.multi_agent_version == Some(MultiAgentVersion::V2)
|
||||
})
|
||||
.unwrap_or(true)
|
||||
{
|
||||
let Some(child_agent_path) = child_agent_path.clone() else {
|
||||
|
||||
@@ -561,13 +561,11 @@ async fn spawn_agent_creates_thread_and_sends_prompt() {
|
||||
async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
|
||||
let harness = AgentControlHarness::new().await;
|
||||
let mut parent_config = harness.config.clone();
|
||||
let _ = parent_config.features.enable(Feature::MultiAgentV2);
|
||||
parent_config.multi_agent_v2.root_agent_usage_hint_text =
|
||||
Some("Parent root guidance.".to_string());
|
||||
parent_config.multi_agent_v2.subagent_usage_hint_text =
|
||||
Some("Parent subagent guidance.".to_string());
|
||||
let mut child_config = harness.config.clone();
|
||||
let _ = child_config.features.enable(Feature::MultiAgentV2);
|
||||
child_config.multi_agent_v2.root_agent_usage_hint_text =
|
||||
Some("Child root guidance.".to_string());
|
||||
child_config.multi_agent_v2.subagent_usage_hint_text =
|
||||
@@ -655,13 +653,14 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
|
||||
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id,
|
||||
depth: 1,
|
||||
agent_path: None,
|
||||
agent_path: Some(AgentPath::try_from("/root/worker").expect("agent path")),
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
})),
|
||||
SpawnAgentOptions {
|
||||
fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()),
|
||||
fork_mode: Some(SpawnAgentForkMode::FullHistory),
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
@@ -814,13 +813,14 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() {
|
||||
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id,
|
||||
depth: 1,
|
||||
agent_path: None,
|
||||
agent_path: Some(AgentPath::try_from("/root/worker").expect("agent path")),
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
})),
|
||||
SpawnAgentOptions {
|
||||
fork_parent_spawn_call_id: Some(parent_spawn_call_id),
|
||||
fork_mode: Some(SpawnAgentForkMode::FullHistory),
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
@@ -1119,6 +1119,7 @@ async fn spawn_agent_fork_last_n_turns_drops_parent_startup_prefix_when_under_li
|
||||
SpawnAgentOptions {
|
||||
fork_parent_spawn_call_id: Some(parent_spawn_call_id),
|
||||
fork_mode: Some(SpawnAgentForkMode::LastNTurns(2)),
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
@@ -1222,13 +1223,14 @@ async fn spawn_agent_fork_last_n_turns_strips_parent_usage_hints() {
|
||||
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id,
|
||||
depth: 1,
|
||||
agent_path: None,
|
||||
agent_path: Some(AgentPath::try_from("/root/worker").expect("agent path")),
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
})),
|
||||
SpawnAgentOptions {
|
||||
fork_parent_spawn_call_id: Some(parent_spawn_call_id),
|
||||
fork_mode: Some(SpawnAgentForkMode::LastNTurns(2)),
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
@@ -1527,12 +1529,11 @@ async fn spawn_child_completion_notifies_parent_history() {
|
||||
async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
|
||||
let harness = AgentControlHarness::new().await;
|
||||
let (root_thread_id, root_thread) = harness.start_thread().await;
|
||||
let mut config = harness.config.clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
let config = harness.config.clone();
|
||||
let worker_path = AgentPath::root().join("worker_a").expect("worker path");
|
||||
let worker_thread_id = harness
|
||||
.control
|
||||
.spawn_agent(
|
||||
.spawn_agent_with_metadata(
|
||||
config.clone(),
|
||||
text_input("hello worker"),
|
||||
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
@@ -1542,13 +1543,18 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
|
||||
agent_nickname: None,
|
||||
agent_role: Some("explorer".to_string()),
|
||||
})),
|
||||
SpawnAgentOptions {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("worker spawn should succeed");
|
||||
.expect("worker spawn should succeed")
|
||||
.thread_id;
|
||||
let tester_path = worker_path.join("tester").expect("tester path");
|
||||
let tester_thread_id = harness
|
||||
.control
|
||||
.spawn_agent(
|
||||
.spawn_agent_with_metadata(
|
||||
config,
|
||||
text_input("hello tester"),
|
||||
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
@@ -1558,9 +1564,14 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
|
||||
agent_nickname: None,
|
||||
agent_role: Some("explorer".to_string()),
|
||||
})),
|
||||
SpawnAgentOptions {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("tester spawn should succeed");
|
||||
.expect("tester spawn should succeed")
|
||||
.thread_id;
|
||||
harness
|
||||
.control
|
||||
.shutdown_live_agent(worker_thread_id)
|
||||
@@ -2601,6 +2612,79 @@ async fn resume_closed_child_reopens_open_descendants() {
|
||||
.expect("parent shutdown should succeed");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resume_agent_from_rollout_restores_persisted_multi_agent_version() {
|
||||
let harness = AgentControlHarness::new().await;
|
||||
let (parent_thread_id, _parent_thread) = harness.start_thread().await;
|
||||
let child_thread_id = harness
|
||||
.control
|
||||
.spawn_agent_with_metadata(
|
||||
harness.config.clone(),
|
||||
text_input("hello"),
|
||||
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id,
|
||||
depth: 1,
|
||||
agent_path: None,
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
})),
|
||||
SpawnAgentOptions {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("child spawn should succeed")
|
||||
.thread_id;
|
||||
let child_thread = harness
|
||||
.manager
|
||||
.get_thread(child_thread_id)
|
||||
.await
|
||||
.expect("child thread should exist");
|
||||
persist_thread_for_tree_resume(&child_thread, "persist selector").await;
|
||||
assert_eq!(
|
||||
child_thread.codex.session.multi_agent_version,
|
||||
Some(MultiAgentVersion::V2)
|
||||
);
|
||||
|
||||
let _ = harness
|
||||
.control
|
||||
.shutdown_live_agent(child_thread_id)
|
||||
.await
|
||||
.expect("child shutdown should submit");
|
||||
let resumed_thread_id = harness
|
||||
.control
|
||||
.resume_agent_from_rollout(
|
||||
harness.config.clone(),
|
||||
child_thread_id,
|
||||
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id,
|
||||
depth: 1,
|
||||
agent_path: None,
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.expect("child resume should succeed");
|
||||
let resumed_thread = harness
|
||||
.manager
|
||||
.get_thread(resumed_thread_id)
|
||||
.await
|
||||
.expect("resumed child thread should exist");
|
||||
|
||||
assert_eq!(
|
||||
resumed_thread.codex.session.multi_agent_version,
|
||||
Some(MultiAgentVersion::V2)
|
||||
);
|
||||
|
||||
let _ = harness
|
||||
.control
|
||||
.shutdown_agent_tree(parent_thread_id)
|
||||
.await
|
||||
.expect("agent tree shutdown should succeed");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resume_agent_from_rollout_reopens_open_descendants_after_manager_shutdown() {
|
||||
let harness = AgentControlHarness::new().await;
|
||||
|
||||
@@ -77,6 +77,7 @@ pub(crate) async fn run_codex_thread_interactive(
|
||||
let CodexSpawnOk { codex, .. } = Box::pin(Codex::spawn(CodexSpawnArgs {
|
||||
config,
|
||||
installation_id: parent_session.installation_id.clone(),
|
||||
parent_multi_agent_version: parent_ctx.multi_agent_version,
|
||||
auth_manager,
|
||||
models_manager,
|
||||
environment_manager: Arc::clone(&parent_session.services.environment_manager),
|
||||
|
||||
@@ -97,6 +97,7 @@ use codex_protocol::models::SandboxEnforcement;
|
||||
use codex_protocol::models::format_allow_prefixes;
|
||||
use codex_protocol::openai_models::ModelInfo;
|
||||
use codex_protocol::openai_models::ModelPreset;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::permissions::FileSystemSandboxPolicy;
|
||||
use codex_protocol::permissions::NetworkSandboxPolicy;
|
||||
use codex_protocol::protocol::AdditionalContextEntry;
|
||||
@@ -225,6 +226,7 @@ use self::turn::collect_explicit_app_ids_from_skill_items;
|
||||
use self::turn::realtime_text_for_event;
|
||||
use self::turn_context::TurnContext;
|
||||
use self::turn_context::TurnSkillsContext;
|
||||
use self::turn_context::resolve_multi_agent_version;
|
||||
#[cfg(test)]
|
||||
mod rollout_reconstruction_tests;
|
||||
|
||||
@@ -392,6 +394,7 @@ pub struct CodexSpawnOk {
|
||||
pub(crate) struct CodexSpawnArgs {
|
||||
pub(crate) config: Config,
|
||||
pub(crate) installation_id: String,
|
||||
pub(crate) parent_multi_agent_version: Option<MultiAgentVersion>,
|
||||
pub(crate) auth_manager: Arc<AuthManager>,
|
||||
pub(crate) models_manager: SharedModelsManager,
|
||||
pub(crate) environment_manager: Arc<EnvironmentManager>,
|
||||
@@ -457,6 +460,7 @@ impl Codex {
|
||||
let CodexSpawnArgs {
|
||||
mut config,
|
||||
installation_id,
|
||||
parent_multi_agent_version,
|
||||
auth_manager,
|
||||
models_manager,
|
||||
environment_manager,
|
||||
@@ -485,14 +489,6 @@ impl Codex {
|
||||
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
|
||||
let (tx_event, rx_event) = async_channel::unbounded();
|
||||
|
||||
if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = session_source
|
||||
&& depth >= config.agent_max_depth
|
||||
&& !config.features.enabled(Feature::MultiAgentV2)
|
||||
{
|
||||
let _ = config.features.disable(Feature::SpawnCsv);
|
||||
let _ = config.features.disable(Feature::Collab);
|
||||
}
|
||||
|
||||
let primary_environment = environment_selections.primary_environment();
|
||||
let mut user_instruction_warnings = Vec::new();
|
||||
let user_instructions = AgentsMdManager::new(&config)
|
||||
@@ -518,7 +514,6 @@ impl Codex {
|
||||
)
|
||||
};
|
||||
|
||||
let config = Arc::new(config);
|
||||
let refresh_strategy = if session_source.is_non_root_agent() {
|
||||
codex_models_manager::manager::RefreshStrategy::Offline
|
||||
} else {
|
||||
@@ -543,6 +538,21 @@ impl Codex {
|
||||
let model_info = models_manager
|
||||
.get_model_info(model.as_str(), &config.to_models_manager_config())
|
||||
.await;
|
||||
let multi_agent_version = resolve_multi_agent_version(
|
||||
&config,
|
||||
&session_source,
|
||||
conversation_history
|
||||
.get_multi_agent_version()
|
||||
.or(parent_multi_agent_version),
|
||||
);
|
||||
if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = session_source
|
||||
&& depth >= config.agent_max_depth
|
||||
&& multi_agent_version != Some(MultiAgentVersion::V2)
|
||||
{
|
||||
let _ = config.features.disable(Feature::SpawnCsv);
|
||||
let _ = config.features.disable(Feature::Collab);
|
||||
}
|
||||
let config = Arc::new(config);
|
||||
let base_instructions = config
|
||||
.base_instructions
|
||||
.clone()
|
||||
@@ -610,6 +620,7 @@ impl Codex {
|
||||
session_configuration,
|
||||
config.clone(),
|
||||
installation_id,
|
||||
multi_agent_version,
|
||||
auth_manager.clone(),
|
||||
models_manager.clone(),
|
||||
exec_policy,
|
||||
@@ -1644,13 +1655,13 @@ impl Session {
|
||||
}
|
||||
}
|
||||
|
||||
/// Forwards terminal turn events from spawned MultiAgentV2 children to their direct parent.
|
||||
/// Forwards terminal turn events from children spawned by a MultiAgentV2 parent.
|
||||
async fn maybe_notify_parent_of_terminal_turn(
|
||||
&self,
|
||||
turn_context: &TurnContext,
|
||||
msg: &EventMsg,
|
||||
) {
|
||||
if !self.enabled(Feature::MultiAgentV2) {
|
||||
if turn_context.multi_agent_version != Some(MultiAgentVersion::V2) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
use crate::session::turn_context::TurnContext;
|
||||
use codex_features::Feature;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
|
||||
@@ -7,7 +7,7 @@ pub(super) fn usage_hint_text<'a>(
|
||||
turn_context: &'a TurnContext,
|
||||
session_source: &SessionSource,
|
||||
) -> Option<&'a str> {
|
||||
if !turn_context.features.enabled(Feature::MultiAgentV2) {
|
||||
if turn_context.multi_agent_version != Some(MultiAgentVersion::V2) {
|
||||
return None;
|
||||
}
|
||||
|
||||
|
||||
@@ -108,6 +108,7 @@ pub(super) async fn spawn_review_thread(
|
||||
auth_manager: auth_manager_for_context,
|
||||
model_info: model_info.clone(),
|
||||
tool_mode,
|
||||
multi_agent_version: None,
|
||||
session_telemetry: session_telemetry_for_context,
|
||||
provider: provider_for_context,
|
||||
reasoning_effort,
|
||||
|
||||
@@ -7,10 +7,14 @@ use crate::state::ActiveTurn;
|
||||
use codex_protocol::SessionId;
|
||||
use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE;
|
||||
use codex_protocol::config_types::ServiceTier;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::permissions::FileSystemPath;
|
||||
use codex_protocol::permissions::FileSystemSpecialPath;
|
||||
use codex_protocol::protocol::ThreadSource;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use std::fs::FileTimes;
|
||||
use std::fs::OpenOptions;
|
||||
use std::fs::metadata;
|
||||
use tokio::sync::Semaphore;
|
||||
|
||||
/// Context for an initialized model agent
|
||||
@@ -19,6 +23,8 @@ use tokio::sync::Semaphore;
|
||||
pub(crate) struct Session {
|
||||
pub(crate) conversation_id: ThreadId,
|
||||
pub(crate) installation_id: String,
|
||||
/// Thread-scoped multi-agent runtime selector, resolved when the thread starts.
|
||||
pub(crate) multi_agent_version: Option<MultiAgentVersion>,
|
||||
pub(super) tx_event: Sender<Event>,
|
||||
pub(super) agent_status: watch::Sender<AgentStatus>,
|
||||
pub(super) out_of_band_elicitation_paused: watch::Sender<bool>,
|
||||
@@ -484,6 +490,7 @@ impl Session {
|
||||
mut session_configuration: SessionConfiguration,
|
||||
config: Arc<Config>,
|
||||
installation_id: String,
|
||||
multi_agent_version: Option<MultiAgentVersion>,
|
||||
auth_manager: Arc<AuthManager>,
|
||||
models_manager: SharedModelsManager,
|
||||
exec_policy: Arc<ExecPolicyManager>,
|
||||
@@ -559,7 +566,7 @@ impl Session {
|
||||
metadata: ThreadPersistenceMetadata {
|
||||
cwd: Some(config.cwd.to_path_buf()),
|
||||
model_provider: config.model_provider_id.clone(),
|
||||
multi_agent_version: None,
|
||||
multi_agent_version,
|
||||
memory_mode: if config.memories.generate_memories {
|
||||
ThreadMemoryMode::Enabled
|
||||
} else {
|
||||
@@ -572,7 +579,7 @@ impl Session {
|
||||
.await?
|
||||
}
|
||||
InitialHistory::Resumed(resumed_history) => {
|
||||
LiveThread::resume(
|
||||
let live_thread = LiveThread::resume(
|
||||
Arc::clone(&thread_store),
|
||||
ResumeThreadParams {
|
||||
thread_id: resumed_history.conversation_id,
|
||||
@@ -582,7 +589,7 @@ impl Session {
|
||||
metadata: ThreadPersistenceMetadata {
|
||||
cwd: Some(config.cwd.to_path_buf()),
|
||||
model_provider: config.model_provider_id.clone(),
|
||||
multi_agent_version: None,
|
||||
multi_agent_version,
|
||||
memory_mode: if config.memories.generate_memories {
|
||||
ThreadMemoryMode::Enabled
|
||||
} else {
|
||||
@@ -592,7 +599,41 @@ impl Session {
|
||||
event_persistence_mode,
|
||||
},
|
||||
)
|
||||
.await?
|
||||
.await?;
|
||||
if initial_history.get_multi_agent_version().is_none()
|
||||
&& let Some(multi_agent_version) = multi_agent_version
|
||||
&& let Some(mut session_meta) = resumed_history
|
||||
.history
|
||||
.iter()
|
||||
.rev()
|
||||
.find_map(|item| match item {
|
||||
RolloutItem::SessionMeta(meta_line)
|
||||
if meta_line.meta.id == resumed_history.conversation_id =>
|
||||
{
|
||||
Some(meta_line.clone())
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
{
|
||||
session_meta.meta.multi_agent_version = Some(multi_agent_version);
|
||||
let rollout_modified_at = if let Some(rollout_path) =
|
||||
live_thread.local_rollout_path().await?
|
||||
{
|
||||
Some((rollout_path.clone(), metadata(rollout_path)?.modified()?))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
live_thread
|
||||
.append_items(&[RolloutItem::SessionMeta(session_meta)])
|
||||
.await?;
|
||||
if let Some((rollout_path, modified_at)) = rollout_modified_at {
|
||||
OpenOptions::new()
|
||||
.append(true)
|
||||
.open(rollout_path)?
|
||||
.set_times(FileTimes::new().set_modified(modified_at))?;
|
||||
}
|
||||
}
|
||||
live_thread
|
||||
}
|
||||
};
|
||||
Ok(Some(live_thread))
|
||||
@@ -1057,6 +1098,7 @@ impl Session {
|
||||
let sess = Arc::new(Session {
|
||||
conversation_id: thread_id,
|
||||
installation_id,
|
||||
multi_agent_version,
|
||||
tx_event: tx_event.clone(),
|
||||
agent_status,
|
||||
out_of_band_elicitation_paused,
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use super::turn_context::TurnEnvironment;
|
||||
use super::turn_context::resolve_multi_agent_version;
|
||||
use super::*;
|
||||
use crate::config::ConfigBuilder;
|
||||
use crate::config::ConfigOverrides;
|
||||
@@ -44,6 +45,7 @@ use codex_protocol::models::FunctionCallOutputPayload;
|
||||
use codex_protocol::models::PermissionProfile;
|
||||
use codex_protocol::models::SandboxEnforcement;
|
||||
use codex_protocol::openai_models::ModelServiceTier;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::permissions::FileSystemAccessMode;
|
||||
use codex_protocol::permissions::FileSystemPath;
|
||||
use codex_protocol::permissions::FileSystemSandboxEntry;
|
||||
@@ -4493,6 +4495,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() {
|
||||
session_configuration,
|
||||
Arc::clone(&config),
|
||||
"11111111-1111-4111-8111-111111111111".to_string(),
|
||||
/*multi_agent_version*/ None,
|
||||
auth_manager,
|
||||
models_manager,
|
||||
Arc::new(ExecPolicyManager::default()),
|
||||
@@ -4722,11 +4725,13 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
|
||||
"turn_id".to_string(),
|
||||
skills_outcome,
|
||||
/*goal_tools_supported*/ true,
|
||||
/*multi_agent_version*/ None,
|
||||
);
|
||||
|
||||
let session = Session {
|
||||
conversation_id: thread_id,
|
||||
installation_id: "11111111-1111-4111-8111-111111111111".to_string(),
|
||||
multi_agent_version: None,
|
||||
tx_event,
|
||||
agent_status: agent_status_tx,
|
||||
out_of_band_elicitation_paused: watch::channel(false).0,
|
||||
@@ -4839,6 +4844,7 @@ async fn make_session_with_config_and_rx(
|
||||
session_configuration,
|
||||
Arc::clone(&config),
|
||||
"11111111-1111-4111-8111-111111111111".to_string(),
|
||||
/*multi_agent_version*/ None,
|
||||
auth_manager,
|
||||
models_manager,
|
||||
Arc::new(ExecPolicyManager::default()),
|
||||
@@ -4943,6 +4949,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx(
|
||||
session_configuration,
|
||||
Arc::clone(&config),
|
||||
"11111111-1111-4111-8111-111111111111".to_string(),
|
||||
/*multi_agent_version*/ None,
|
||||
auth_manager,
|
||||
models_manager,
|
||||
Arc::new(ExecPolicyManager::default()),
|
||||
@@ -6566,11 +6573,13 @@ where
|
||||
"turn_id".to_string(),
|
||||
skills_outcome,
|
||||
/*goal_tools_supported*/ true,
|
||||
/*multi_agent_version*/ None,
|
||||
));
|
||||
|
||||
let session = Arc::new(Session {
|
||||
conversation_id: thread_id,
|
||||
installation_id: "11111111-1111-4111-8111-111111111111".to_string(),
|
||||
multi_agent_version: None,
|
||||
tx_event,
|
||||
agent_status: agent_status_tx,
|
||||
out_of_band_elicitation_paused: watch::channel(false).0,
|
||||
@@ -7009,18 +7018,28 @@ async fn build_initial_context_uses_previous_realtime_state() {
|
||||
async fn make_multi_agent_v2_usage_hint_test_session(
|
||||
enable_multi_agent_v2: bool,
|
||||
) -> (Arc<Session>, Arc<TurnContext>) {
|
||||
let (session, turn_context, _rx_event) = make_session_and_context_with_auth_and_config_and_rx(
|
||||
CodexAuth::from_api_key("Test API Key"),
|
||||
Vec::new(),
|
||||
|config| {
|
||||
if enable_multi_agent_v2 {
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
}
|
||||
config.multi_agent_v2.root_agent_usage_hint_text = Some("Root guidance.".to_string());
|
||||
config.multi_agent_v2.subagent_usage_hint_text = Some("Subagent guidance.".to_string());
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let (mut session, mut turn_context, _rx_event) =
|
||||
make_session_and_context_with_auth_and_config_and_rx(
|
||||
CodexAuth::from_api_key("Test API Key"),
|
||||
Vec::new(),
|
||||
|config| {
|
||||
if enable_multi_agent_v2 {
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
}
|
||||
config.multi_agent_v2.root_agent_usage_hint_text =
|
||||
Some("Root guidance.".to_string());
|
||||
config.multi_agent_v2.subagent_usage_hint_text =
|
||||
Some("Subagent guidance.".to_string());
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let multi_agent_version = enable_multi_agent_v2.then_some(MultiAgentVersion::V2);
|
||||
Arc::get_mut(&mut session)
|
||||
.expect("session should not be shared")
|
||||
.multi_agent_version = multi_agent_version;
|
||||
Arc::get_mut(&mut turn_context)
|
||||
.expect("turn context should not be shared")
|
||||
.multi_agent_version = multi_agent_version;
|
||||
(session, turn_context)
|
||||
}
|
||||
|
||||
@@ -7174,6 +7193,93 @@ async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_feature_dis
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_initial_context_omits_multi_agent_v2_usage_hints_when_selector_is_v1() {
|
||||
let (session, mut turn_context) =
|
||||
make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ true).await;
|
||||
Arc::get_mut(&mut turn_context)
|
||||
.expect("turn context should not be shared")
|
||||
.multi_agent_version = Some(MultiAgentVersion::V1);
|
||||
|
||||
let initial_context = session.build_initial_context(turn_context.as_ref()).await;
|
||||
|
||||
let developer_messages = developer_message_texts(&initial_context);
|
||||
assert!(
|
||||
!developer_messages.iter().any(|message| {
|
||||
matches!(
|
||||
message.as_slice(),
|
||||
["Root guidance."] | ["Subagent guidance."]
|
||||
)
|
||||
}),
|
||||
"did not expect multi-agent v2 usage hint developer messages, got {developer_messages:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_initial_context_adds_multi_agent_v2_usage_hint_when_selector_is_v2() {
|
||||
let (session, mut turn_context) =
|
||||
make_multi_agent_v2_usage_hint_test_session(/*enable_multi_agent_v2*/ false).await;
|
||||
Arc::get_mut(&mut turn_context)
|
||||
.expect("turn context should not be shared")
|
||||
.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let initial_context = session.build_initial_context(turn_context.as_ref()).await;
|
||||
|
||||
let developer_messages = developer_message_texts(&initial_context);
|
||||
assert!(
|
||||
developer_messages
|
||||
.iter()
|
||||
.any(|message| message.as_slice() == ["Root guidance."]),
|
||||
"expected standalone root usage hint developer message, got {developer_messages:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn spawned_child_multi_agent_version_follows_parent_system() {
|
||||
let (_session, turn_context) = make_session_and_context().await;
|
||||
let child_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id: ThreadId::new(),
|
||||
depth: 1,
|
||||
agent_path: None,
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
});
|
||||
let resolved_from_v2_parent = resolve_multi_agent_version(
|
||||
turn_context.config.as_ref(),
|
||||
&child_source,
|
||||
Some(MultiAgentVersion::V2),
|
||||
);
|
||||
|
||||
let resolved_from_v1_parent = resolve_multi_agent_version(
|
||||
turn_context.config.as_ref(),
|
||||
&child_source,
|
||||
Some(MultiAgentVersion::V1),
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
(resolved_from_v2_parent, resolved_from_v1_parent),
|
||||
(Some(MultiAgentVersion::V2), Some(MultiAgentVersion::V1),)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn model_switch_preserves_thread_multi_agent_version() {
|
||||
let (session, mut turn_context) = make_session_and_context().await;
|
||||
turn_context.multi_agent_version = Some(MultiAgentVersion::V1);
|
||||
|
||||
let switched_context = turn_context
|
||||
.with_model(
|
||||
turn_context.model_info.slug.clone(),
|
||||
&session.services.models_manager,
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(
|
||||
switched_context.multi_agent_version,
|
||||
Some(MultiAgentVersion::V1)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_initial_context_omits_default_image_save_location_with_image_history() {
|
||||
let (session, turn_context) = make_session_and_context().await;
|
||||
|
||||
@@ -681,6 +681,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
|
||||
let CodexSpawnOk { codex, .. } = Codex::spawn(CodexSpawnArgs {
|
||||
config,
|
||||
installation_id: "11111111-1111-4111-8111-111111111111".to_string(),
|
||||
parent_multi_agent_version: None,
|
||||
auth_manager,
|
||||
models_manager,
|
||||
environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
|
||||
|
||||
@@ -2,11 +2,16 @@ use super::*;
|
||||
use crate::SkillLoadOutcome;
|
||||
use crate::config::GhostSnapshotConfig;
|
||||
use crate::environment_selection::ResolvedTurnEnvironments;
|
||||
use crate::guardian::is_guardian_reviewer_source;
|
||||
use codex_model_provider::SharedModelProvider;
|
||||
use codex_model_provider::create_model_provider;
|
||||
use codex_protocol::SessionId;
|
||||
use codex_protocol::models::AdditionalPermissionProfile;
|
||||
use codex_protocol::openai_models::ModelInfo;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::openai_models::ToolMode;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
use codex_protocol::protocol::ThreadSource;
|
||||
use codex_protocol::protocol::TurnEnvironmentSelection;
|
||||
use codex_sandboxing::compatibility_sandbox_policy_for_permission_profile;
|
||||
@@ -57,6 +62,7 @@ pub struct TurnContext {
|
||||
pub(crate) auth_manager: Option<Arc<AuthManager>>,
|
||||
pub(crate) model_info: ModelInfo,
|
||||
pub(crate) tool_mode: ToolMode,
|
||||
pub(crate) multi_agent_version: Option<MultiAgentVersion>,
|
||||
pub(crate) session_telemetry: SessionTelemetry,
|
||||
pub(crate) provider: SharedModelProvider,
|
||||
pub(crate) reasoning_effort: Option<ReasoningEffortConfig>,
|
||||
@@ -99,6 +105,31 @@ pub struct TurnContext {
|
||||
pub(crate) server_model_warning_emitted: AtomicBool,
|
||||
pub(crate) model_verification_emitted: AtomicBool,
|
||||
}
|
||||
|
||||
pub(crate) fn resolve_multi_agent_version(
|
||||
config: &Config,
|
||||
session_source: &SessionSource,
|
||||
inherited_multi_agent_version: Option<MultiAgentVersion>,
|
||||
) -> Option<MultiAgentVersion> {
|
||||
if is_guardian_reviewer_source(session_source)
|
||||
|| matches!(
|
||||
session_source,
|
||||
SessionSource::SubAgent(SubAgentSource::Review)
|
||||
)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
inherited_multi_agent_version.or_else(|| {
|
||||
if config.features.enabled(Feature::MultiAgentV2) {
|
||||
Some(MultiAgentVersion::V2)
|
||||
} else if config.features.enabled(Feature::Collab) {
|
||||
Some(MultiAgentVersion::V1)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
impl TurnContext {
|
||||
pub(crate) fn permission_profile(&self) -> PermissionProfile {
|
||||
self.permission_profile.clone()
|
||||
@@ -183,6 +214,7 @@ impl TurnContext {
|
||||
ToolMode::Direct
|
||||
}
|
||||
});
|
||||
let multi_agent_version = self.multi_agent_version;
|
||||
let truncation_policy = model_info.truncation_policy.into();
|
||||
let supported_reasoning_levels = model_info
|
||||
.supported_reasoning_levels
|
||||
@@ -224,6 +256,7 @@ impl TurnContext {
|
||||
auth_manager: self.auth_manager.clone(),
|
||||
model_info: model_info.clone(),
|
||||
tool_mode,
|
||||
multi_agent_version,
|
||||
session_telemetry: self
|
||||
.session_telemetry
|
||||
.clone()
|
||||
@@ -463,6 +496,7 @@ impl Session {
|
||||
sub_id: String,
|
||||
skills_outcome: Arc<SkillLoadOutcome>,
|
||||
goal_tools_supported: bool,
|
||||
multi_agent_version: Option<MultiAgentVersion>,
|
||||
) -> TurnContext {
|
||||
let reasoning_effort = session_configuration.collaboration_mode.reasoning_effort();
|
||||
let reasoning_summary = session_configuration
|
||||
@@ -524,6 +558,7 @@ impl Session {
|
||||
auth_manager: auth_manager_for_context,
|
||||
model_info: model_info.clone(),
|
||||
tool_mode,
|
||||
multi_agent_version,
|
||||
session_telemetry: session_telemetry_for_context,
|
||||
provider: provider_for_context,
|
||||
reasoning_effort,
|
||||
@@ -745,6 +780,7 @@ impl Session {
|
||||
sub_id,
|
||||
skills_outcome,
|
||||
goal_tools_supported,
|
||||
self.multi_agent_version,
|
||||
);
|
||||
turn_context.realtime_active = self.conversation.running_state().await.is_some();
|
||||
|
||||
|
||||
@@ -34,6 +34,7 @@ use crate::state::ActiveTurn;
|
||||
use crate::state::RunningTask;
|
||||
use crate::state::TaskKind;
|
||||
use codex_analytics::TurnTokenUsageFact;
|
||||
use codex_features::Feature;
|
||||
use codex_login::AuthManager;
|
||||
use codex_models_manager::manager::SharedModelsManager;
|
||||
use codex_otel::SessionTelemetry;
|
||||
@@ -50,8 +51,8 @@ use codex_protocol::protocol::TurnAbortedEvent;
|
||||
use codex_protocol::protocol::TurnCompleteEvent;
|
||||
use codex_protocol::protocol::WarningEvent;
|
||||
|
||||
use codex_features::Feature;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
pub(crate) use compact::CompactTask;
|
||||
pub(crate) use regular::RegularTask;
|
||||
pub(crate) use review::ReviewTask;
|
||||
@@ -70,11 +71,14 @@ pub(crate) enum InterruptedTurnHistoryMarker {
|
||||
}
|
||||
|
||||
impl InterruptedTurnHistoryMarker {
|
||||
pub(crate) fn from_config(config: &Config) -> Self {
|
||||
pub(crate) fn from_config(
|
||||
config: &Config,
|
||||
multi_agent_version: Option<MultiAgentVersion>,
|
||||
) -> Self {
|
||||
if !config.agent_interrupt_message_enabled {
|
||||
return Self::Disabled;
|
||||
}
|
||||
if config.features.enabled(Feature::MultiAgentV2) {
|
||||
if multi_agent_version == Some(MultiAgentVersion::V2) {
|
||||
Self::Developer
|
||||
} else {
|
||||
Self::ContextualUser
|
||||
@@ -840,9 +844,11 @@ impl Session {
|
||||
.await;
|
||||
|
||||
if reason == TurnAbortReason::Interrupted
|
||||
&& let Some(marker) = interrupted_turn_history_marker(
|
||||
InterruptedTurnHistoryMarker::from_config(task.turn_context.config.as_ref()),
|
||||
)
|
||||
&& let Some(marker) =
|
||||
interrupted_turn_history_marker(InterruptedTurnHistoryMarker::from_config(
|
||||
task.turn_context.config.as_ref(),
|
||||
task.turn_context.multi_agent_version,
|
||||
))
|
||||
{
|
||||
self.record_conversation_items(
|
||||
task.turn_context.as_ref(),
|
||||
|
||||
@@ -12,6 +12,7 @@ use crate::session::Codex;
|
||||
use crate::session::CodexSpawnArgs;
|
||||
use crate::session::CodexSpawnOk;
|
||||
use crate::session::INITIAL_SUBMIT_ID;
|
||||
use crate::session::turn_context::resolve_multi_agent_version;
|
||||
use crate::shell_snapshot::ShellSnapshot;
|
||||
use crate::tasks::InterruptedTurnHistoryMarker;
|
||||
use crate::tasks::interrupted_turn_history_marker;
|
||||
@@ -34,6 +35,7 @@ use codex_protocol::config_types::CollaborationModeMask;
|
||||
use codex_protocol::error::CodexErr;
|
||||
use codex_protocol::error::Result as CodexResult;
|
||||
use codex_protocol::openai_models::ModelPreset;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::protocol::Event;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::InitialHistory;
|
||||
@@ -595,12 +597,14 @@ impl ThreadManager {
|
||||
.unwrap_or_else(|| (self.state.session_source.clone(), None));
|
||||
let session_source = options.session_source.unwrap_or(resumed_session_source);
|
||||
let thread_source = options.thread_source.or(resumed_thread_source);
|
||||
let parent_multi_agent_version = options.initial_history.get_multi_agent_version();
|
||||
Box::pin(self.state.spawn_thread_with_source(
|
||||
options.config,
|
||||
options.initial_history,
|
||||
Arc::clone(&self.state.auth_manager),
|
||||
self.agent_control(),
|
||||
session_source,
|
||||
parent_multi_agent_version,
|
||||
forked_from_thread_id,
|
||||
thread_source,
|
||||
options.dynamic_tools,
|
||||
@@ -640,7 +644,10 @@ impl ThreadManager {
|
||||
options.initial_history = fork_history_from_snapshot(
|
||||
ForkSnapshot::Interrupted,
|
||||
history,
|
||||
InterruptedTurnHistoryMarker::from_config(&options.config),
|
||||
InterruptedTurnHistoryMarker::from_config(
|
||||
&options.config,
|
||||
fork_source.codex.session.multi_agent_version,
|
||||
),
|
||||
);
|
||||
self.start_thread_with_options_and_fork_source(options, Some(forked_from_thread_id))
|
||||
.await
|
||||
@@ -685,6 +692,7 @@ impl ThreadManager {
|
||||
auth_manager,
|
||||
self.agent_control(),
|
||||
session_source,
|
||||
/*parent_multi_agent_version*/ None,
|
||||
/*forked_from_thread_id*/ None,
|
||||
thread_source,
|
||||
Vec::new(),
|
||||
@@ -746,6 +754,7 @@ impl ThreadManager {
|
||||
auth_manager,
|
||||
self.agent_control(),
|
||||
session_source,
|
||||
/*parent_multi_agent_version*/ None,
|
||||
/*forked_from_thread_id*/ None,
|
||||
thread_source,
|
||||
Vec::new(),
|
||||
@@ -905,7 +914,17 @@ impl ThreadManager {
|
||||
InitialHistory::Forked(_) => history.forked_from_id(),
|
||||
InitialHistory::New | InitialHistory::Cleared => None,
|
||||
};
|
||||
let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config);
|
||||
let session_source = history
|
||||
.get_resumed_session_sources()
|
||||
.map(|(session_source, _)| session_source)
|
||||
.unwrap_or_else(|| self.state.session_source.clone());
|
||||
let multi_agent_version = resolve_multi_agent_version(
|
||||
&config,
|
||||
&session_source,
|
||||
history.get_multi_agent_version(),
|
||||
);
|
||||
let interrupted_marker =
|
||||
InterruptedTurnHistoryMarker::from_config(&config, multi_agent_version);
|
||||
let history = fork_history_from_snapshot(snapshot, history, interrupted_marker);
|
||||
let environments = default_thread_environment_selections(
|
||||
self.state.environment_manager.as_ref(),
|
||||
@@ -1039,6 +1058,7 @@ impl ThreadManagerState {
|
||||
config,
|
||||
agent_control,
|
||||
self.session_source.clone(),
|
||||
/*parent_multi_agent_version*/ None,
|
||||
/*forked_from_thread_id*/ None,
|
||||
/*thread_source*/ None,
|
||||
/*persist_extended_history*/ false,
|
||||
@@ -1056,6 +1076,7 @@ impl ThreadManagerState {
|
||||
config: Config,
|
||||
agent_control: AgentControl,
|
||||
session_source: SessionSource,
|
||||
parent_multi_agent_version: Option<MultiAgentVersion>,
|
||||
forked_from_thread_id: Option<ThreadId>,
|
||||
thread_source: Option<ThreadSource>,
|
||||
persist_extended_history: bool,
|
||||
@@ -1073,6 +1094,7 @@ impl ThreadManagerState {
|
||||
Arc::clone(&self.auth_manager),
|
||||
agent_control,
|
||||
session_source,
|
||||
parent_multi_agent_version,
|
||||
forked_from_thread_id,
|
||||
thread_source,
|
||||
Vec::new(),
|
||||
@@ -1108,6 +1130,7 @@ impl ThreadManagerState {
|
||||
Arc::clone(&self.auth_manager),
|
||||
agent_control,
|
||||
session_source,
|
||||
/*parent_multi_agent_version*/ None,
|
||||
/*forked_from_thread_id*/ None,
|
||||
thread_source,
|
||||
Vec::new(),
|
||||
@@ -1129,6 +1152,7 @@ impl ThreadManagerState {
|
||||
initial_history: InitialHistory,
|
||||
agent_control: AgentControl,
|
||||
session_source: SessionSource,
|
||||
parent_multi_agent_version: Option<MultiAgentVersion>,
|
||||
thread_source: Option<ThreadSource>,
|
||||
forked_from_thread_id: Option<ThreadId>,
|
||||
persist_extended_history: bool,
|
||||
@@ -1145,6 +1169,7 @@ impl ThreadManagerState {
|
||||
Arc::clone(&self.auth_manager),
|
||||
agent_control,
|
||||
session_source,
|
||||
parent_multi_agent_version,
|
||||
forked_from_thread_id,
|
||||
thread_source,
|
||||
Vec::new(),
|
||||
@@ -1176,12 +1201,14 @@ impl ThreadManagerState {
|
||||
environments: Vec<TurnEnvironmentSelection>,
|
||||
user_shell_override: Option<crate::shell::Shell>,
|
||||
) -> CodexResult<NewThread> {
|
||||
let parent_multi_agent_version = initial_history.get_multi_agent_version();
|
||||
Box::pin(self.spawn_thread_with_source(
|
||||
config,
|
||||
initial_history,
|
||||
auth_manager,
|
||||
agent_control,
|
||||
self.session_source.clone(),
|
||||
parent_multi_agent_version,
|
||||
forked_from_thread_id,
|
||||
thread_source,
|
||||
dynamic_tools,
|
||||
@@ -1204,6 +1231,7 @@ impl ThreadManagerState {
|
||||
auth_manager: Arc<AuthManager>,
|
||||
agent_control: AgentControl,
|
||||
session_source: SessionSource,
|
||||
parent_multi_agent_version: Option<MultiAgentVersion>,
|
||||
forked_from_thread_id: Option<ThreadId>,
|
||||
thread_source: Option<ThreadSource>,
|
||||
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
|
||||
@@ -1248,6 +1276,7 @@ impl ThreadManagerState {
|
||||
} = Codex::spawn(CodexSpawnArgs {
|
||||
config,
|
||||
installation_id: self.installation_id.clone(),
|
||||
parent_multi_agent_version,
|
||||
auth_manager,
|
||||
models_manager: Arc::clone(&self.models_manager),
|
||||
environment_manager: Arc::clone(&self.environment_manager),
|
||||
|
||||
@@ -14,6 +14,7 @@ use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::ReasoningItemReasoningSummary;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::openai_models::ModelsResponse;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::protocol::AgentMessageEvent;
|
||||
use codex_protocol::protocol::InitialHistory;
|
||||
use codex_protocol::protocol::InternalSessionSource;
|
||||
@@ -758,6 +759,114 @@ async fn resume_stopped_thread_from_rollout_spawns_new_thread() {
|
||||
.expect("shutdown resumed thread");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_rollout_resume_persists_flag_resolved_multi_agent_version() {
|
||||
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::Collab)
|
||||
.expect("collab should be disableable");
|
||||
config
|
||||
.features
|
||||
.disable(Feature::MultiAgentV2)
|
||||
.expect("multi-agent v2 should be disableable");
|
||||
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 state_db = init_state_db(&config).await.expect("state db");
|
||||
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, Some(state_db.clone())),
|
||||
Some(state_db.clone()),
|
||||
TEST_INSTALLATION_ID.to_string(),
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
|
||||
let source = manager
|
||||
.start_thread(config.clone())
|
||||
.await
|
||||
.expect("start legacy source thread");
|
||||
source.thread.ensure_rollout_materialized().await;
|
||||
source
|
||||
.thread
|
||||
.flush_rollout()
|
||||
.await
|
||||
.expect("flush legacy source rollout");
|
||||
let rollout_path = source
|
||||
.thread
|
||||
.rollout_path()
|
||||
.expect("legacy source rollout path should exist");
|
||||
source
|
||||
.thread
|
||||
.shutdown_and_wait()
|
||||
.await
|
||||
.expect("shutdown legacy source thread");
|
||||
let _ = manager.remove_thread(&source.thread_id).await;
|
||||
|
||||
config
|
||||
.features
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("multi-agent v2 should be enableable");
|
||||
let resumed = manager
|
||||
.resume_thread_from_rollout(
|
||||
config,
|
||||
rollout_path.clone(),
|
||||
auth_manager,
|
||||
/*parent_trace*/ None,
|
||||
)
|
||||
.await
|
||||
.expect("resume legacy source thread");
|
||||
resumed
|
||||
.thread
|
||||
.flush_rollout()
|
||||
.await
|
||||
.expect("flush resumed source rollout");
|
||||
let resumed_history = RolloutRecorder::get_rollout_history(&rollout_path)
|
||||
.await
|
||||
.expect("read resumed source rollout");
|
||||
let persisted_versions = resumed_history
|
||||
.get_rollout_items()
|
||||
.into_iter()
|
||||
.filter_map(|item| match item {
|
||||
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.multi_agent_version),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let metadata = state_db
|
||||
.get_thread(resumed.thread_id)
|
||||
.await
|
||||
.expect("read sqlite metadata")
|
||||
.expect("sqlite metadata");
|
||||
|
||||
assert_eq!(
|
||||
(
|
||||
resumed.thread.codex.session.multi_agent_version,
|
||||
persisted_versions,
|
||||
metadata.multi_agent_version
|
||||
),
|
||||
(
|
||||
Some(MultiAgentVersion::V2),
|
||||
vec![None, Some(MultiAgentVersion::V2)],
|
||||
Some(MultiAgentVersion::V2)
|
||||
)
|
||||
);
|
||||
|
||||
resumed
|
||||
.thread
|
||||
.shutdown_and_wait()
|
||||
.await
|
||||
.expect("shutdown resumed source thread");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resume_stopped_thread_from_rollout_preserves_thread_source() {
|
||||
let temp_dir = tempdir().expect("tempdir");
|
||||
|
||||
@@ -212,6 +212,7 @@ async fn run_agent_job_loop(
|
||||
)))),
|
||||
SpawnAgentOptions {
|
||||
environments: Some(turn.environments.to_selections()),
|
||||
multi_agent_version: turn.multi_agent_version,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
|
||||
@@ -181,7 +181,7 @@ async fn try_resume_closed_agent(
|
||||
receiver_thread_id: ThreadId,
|
||||
child_depth: i32,
|
||||
) -> Result<(), FunctionCallError> {
|
||||
let config = build_agent_resume_config(turn.as_ref(), child_depth)?;
|
||||
let config = build_agent_resume_config(turn.as_ref())?;
|
||||
Box::pin(session.services.agent_control.resume_agent_from_rollout(
|
||||
config,
|
||||
receiver_thread_id,
|
||||
|
||||
@@ -109,7 +109,6 @@ async fn handle_spawn_agent(
|
||||
)
|
||||
.await?;
|
||||
apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?;
|
||||
apply_spawn_agent_overrides(&mut config, child_depth);
|
||||
|
||||
let result = Box::pin(session.services.agent_control.spawn_agent_with_metadata(
|
||||
config,
|
||||
@@ -125,6 +124,7 @@ async fn handle_spawn_agent(
|
||||
fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()),
|
||||
fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory),
|
||||
environments: Some(turn.environments.to_selections()),
|
||||
multi_agent_version: turn.multi_agent_version,
|
||||
},
|
||||
))
|
||||
.await
|
||||
|
||||
@@ -8,7 +8,6 @@ use crate::session::turn_context::TurnContext;
|
||||
use crate::tools::context::FunctionToolOutput;
|
||||
use crate::tools::context::ToolOutput;
|
||||
use crate::tools::context::ToolPayload;
|
||||
use codex_features::Feature;
|
||||
use codex_models_manager::manager::RefreshStrategy;
|
||||
use codex_protocol::AgentPath;
|
||||
use codex_protocol::ThreadId;
|
||||
@@ -211,12 +210,8 @@ pub(crate) fn build_agent_spawn_config(
|
||||
Ok(config)
|
||||
}
|
||||
|
||||
pub(crate) fn build_agent_resume_config(
|
||||
turn: &TurnContext,
|
||||
child_depth: i32,
|
||||
) -> Result<Config, FunctionCallError> {
|
||||
pub(crate) fn build_agent_resume_config(turn: &TurnContext) -> Result<Config, FunctionCallError> {
|
||||
let mut config = build_agent_shared_config(turn)?;
|
||||
apply_spawn_agent_overrides(&mut config, child_depth);
|
||||
// For resume, keep base instructions sourced from rollout/session metadata.
|
||||
config.base_instructions = None;
|
||||
Ok(config)
|
||||
@@ -280,13 +275,6 @@ pub(crate) fn apply_spawn_agent_runtime_overrides(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn apply_spawn_agent_overrides(config: &mut Config, child_depth: i32) {
|
||||
if child_depth >= config.agent_max_depth && !config.features.enabled(Feature::MultiAgentV2) {
|
||||
let _ = config.features.disable(Feature::SpawnCsv);
|
||||
let _ = config.features.disable(Feature::Collab);
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn apply_requested_spawn_agent_model_overrides(
|
||||
session: &Session,
|
||||
turn: &TurnContext,
|
||||
|
||||
@@ -32,6 +32,7 @@ use codex_protocol::models::PermissionProfile;
|
||||
use codex_protocol::models::ResponseInputItem;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::models::SandboxEnforcement;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::openai_models::ReasoningEffort;
|
||||
use codex_protocol::protocol::AgentStatus;
|
||||
use codex_protocol::protocol::AskForApproval;
|
||||
@@ -382,6 +383,7 @@ async fn multi_agent_v2_spawn_fork_turns_all_rejects_agent_type_override() {
|
||||
.expect("test config should allow feature update");
|
||||
let turn = TurnContext {
|
||||
config: Arc::new(config),
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..turn
|
||||
};
|
||||
|
||||
@@ -425,6 +427,7 @@ async fn multi_agent_v2_spawn_defaults_to_full_fork_and_rejects_child_model_over
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let err = SpawnAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -892,6 +895,7 @@ async fn multi_agent_v2_full_history_fork_accepts_explicit_service_tier() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let manager = thread_manager();
|
||||
let root = manager
|
||||
.start_thread((*turn.config).clone())
|
||||
@@ -959,6 +963,7 @@ async fn multi_agent_v2_spawn_partial_fork_turns_allows_agent_type_override() {
|
||||
.expect("test config should allow feature update");
|
||||
let turn = TurnContext {
|
||||
config: Arc::new(config),
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
..turn
|
||||
};
|
||||
|
||||
@@ -1041,6 +1046,7 @@ async fn multi_agent_v2_spawn_requires_task_name() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let invocation = invocation(
|
||||
Arc::new(session),
|
||||
@@ -1075,6 +1081,7 @@ async fn multi_agent_v2_spawn_rejects_legacy_items_field() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let invocation = invocation(
|
||||
Arc::new(session),
|
||||
@@ -1135,6 +1142,7 @@ async fn multi_agent_v2_spawn_returns_path_and_send_message_accepts_relative_pat
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
@@ -1232,6 +1240,7 @@ async fn multi_agent_v2_spawn_rejects_legacy_fork_context() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let err = SpawnAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -1272,6 +1281,7 @@ async fn multi_agent_v2_spawn_rejects_invalid_fork_turns_string() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let err = SpawnAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -1312,6 +1322,7 @@ async fn multi_agent_v2_spawn_rejects_zero_fork_turns() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let err = SpawnAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -1352,6 +1363,7 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let child_path = AgentPath::try_from("/root/worker").expect("agent path");
|
||||
let child_thread_id = session
|
||||
@@ -1428,6 +1440,7 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let child_path = AgentPath::try_from("/root/worker").expect("agent path");
|
||||
let child_thread_id = session
|
||||
@@ -1506,6 +1519,7 @@ async fn multi_agent_v2_list_agents_returns_completed_status_and_last_task_messa
|
||||
let mut config = (*turn.config).clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
@@ -1600,6 +1614,7 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
|
||||
let mut config = (*turn.config).clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config.clone());
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let researcher_path = AgentPath::from_string("/root/researcher".to_string()).expect("path");
|
||||
let worker_path = AgentPath::from_string("/root/researcher/worker".to_string()).expect("path");
|
||||
@@ -1687,6 +1702,7 @@ async fn multi_agent_v2_list_agents_omits_closed_agents() {
|
||||
let mut config = (*turn.config).clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
@@ -1751,6 +1767,7 @@ async fn multi_agent_v2_send_message_rejects_legacy_items_field() {
|
||||
let mut config = turn.config.as_ref().clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -1807,6 +1824,7 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() {
|
||||
let mut config = turn.config.as_ref().clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -1871,15 +1889,16 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() {
|
||||
async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn() {
|
||||
let (mut session, mut turn) = make_session_and_context().await;
|
||||
let manager = thread_manager();
|
||||
let mut config = turn.config.as_ref().clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
let root = manager
|
||||
.start_thread((*turn.config).clone())
|
||||
.start_thread(config.clone())
|
||||
.await
|
||||
.expect("root thread should start");
|
||||
session.services.agent_control = manager.agent_control();
|
||||
session.conversation_id = root.thread_id;
|
||||
let mut config = turn.config.as_ref().clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -2015,6 +2034,7 @@ async fn multi_agent_v2_followup_task_rejects_legacy_items_field() {
|
||||
let mut config = turn.config.as_ref().clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -2068,6 +2088,7 @@ async fn multi_agent_v2_interrupted_turn_does_not_notify_parent() {
|
||||
let mut config = turn.config.as_ref().clone();
|
||||
let _ = config.features.enable(Feature::MultiAgentV2);
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -2148,6 +2169,7 @@ async fn multi_agent_v2_spawn_omits_agent_id_when_named() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let output = SpawnAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -2187,6 +2209,7 @@ async fn multi_agent_v2_spawn_surfaces_task_name_validation_errors() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let invocation = invocation(
|
||||
Arc::new(session),
|
||||
@@ -2397,6 +2420,7 @@ async fn multi_agent_v2_spawn_agent_ignores_configured_max_depth() {
|
||||
session.services.agent_control = manager.agent_control();
|
||||
session.conversation_id = root.thread_id;
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let parent_path = AgentPath::try_from("/root/parent").expect("agent path");
|
||||
turn.session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id: root.thread_id,
|
||||
@@ -2871,6 +2895,7 @@ async fn multi_agent_v2_wait_agent_accepts_timeout_only_argument() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -2956,6 +2981,7 @@ async fn multi_agent_v2_wait_agent_rejects_timeout_below_configured_min() {
|
||||
config.multi_agent_v2.max_wait_timeout_ms = 1_000;
|
||||
config.multi_agent_v2.default_wait_timeout_ms = 50;
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let Err(err) = WaitAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -2986,6 +3012,7 @@ async fn multi_agent_v2_wait_agent_accepts_explicit_timeout_at_configured_min()
|
||||
config.multi_agent_v2.max_wait_timeout_ms = 1_000;
|
||||
config.multi_agent_v2.default_wait_timeout_ms = 50;
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let output = WaitAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -3021,6 +3048,7 @@ async fn multi_agent_v2_wait_agent_uses_configured_default_timeout() {
|
||||
config.multi_agent_v2.max_wait_timeout_ms = 1_000;
|
||||
config.multi_agent_v2.default_wait_timeout_ms = 50;
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -3076,6 +3104,7 @@ async fn multi_agent_v2_wait_agent_allows_zero_configured_timeout() {
|
||||
config.multi_agent_v2.max_wait_timeout_ms = 0;
|
||||
config.multi_agent_v2.default_wait_timeout_ms = 0;
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -3116,6 +3145,7 @@ async fn multi_agent_v2_wait_agent_rejects_timeout_above_configured_max() {
|
||||
config.multi_agent_v2.max_wait_timeout_ms = 50;
|
||||
config.multi_agent_v2.default_wait_timeout_ms = 1;
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let Err(err) = WaitAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -3146,6 +3176,7 @@ async fn multi_agent_v2_wait_agent_accepts_explicit_timeout_at_configured_max()
|
||||
config.multi_agent_v2.max_wait_timeout_ms = 1;
|
||||
config.multi_agent_v2.default_wait_timeout_ms = 1;
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let output = WaitAgentHandlerV2::default()
|
||||
.handle(invocation(
|
||||
@@ -3354,6 +3385,7 @@ async fn multi_agent_v2_wait_agent_returns_summary_for_mailbox_activity() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
@@ -3448,6 +3480,7 @@ async fn multi_agent_v2_wait_agent_returns_for_already_queued_mail() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -3529,6 +3562,7 @@ async fn multi_agent_v2_wait_agent_wakes_on_any_mailbox_notification() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -3620,6 +3654,7 @@ async fn multi_agent_v2_wait_agent_does_not_return_completed_content() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
|
||||
@@ -3709,6 +3744,7 @@ async fn multi_agent_v2_close_agent_accepts_task_name_target() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
@@ -3782,6 +3818,7 @@ async fn multi_agent_v2_close_agent_reaps_stale_task_name_target() {
|
||||
session.services.agent_control = manager.agent_control();
|
||||
session.conversation_id = root.thread_id;
|
||||
turn.config = Arc::new(config.clone());
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
@@ -3888,6 +3925,7 @@ async fn multi_agent_v2_close_agent_rejects_root_target_and_id() {
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
turn.config = Arc::new(config);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
|
||||
let session = Arc::new(session);
|
||||
let turn = Arc::new(turn);
|
||||
@@ -4282,7 +4320,7 @@ async fn build_agent_resume_config_clears_base_instructions() {
|
||||
.set(AskForApproval::OnRequest)
|
||||
.expect("approval policy set");
|
||||
|
||||
let config = build_agent_resume_config(&turn, /*child_depth*/ 0).expect("resume config");
|
||||
let config = build_agent_resume_config(&turn).expect("resume config");
|
||||
|
||||
let mut expected = (*turn.config).clone();
|
||||
expected.base_instructions = None;
|
||||
|
||||
@@ -108,7 +108,6 @@ async fn handle_spawn_agent(
|
||||
)
|
||||
.await?;
|
||||
apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?;
|
||||
apply_spawn_agent_overrides(&mut config, child_depth);
|
||||
|
||||
let spawn_source = thread_spawn_source(
|
||||
session.conversation_id,
|
||||
@@ -145,6 +144,7 @@ async fn handle_spawn_agent(
|
||||
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
|
||||
fork_mode,
|
||||
environments: Some(turn.environments.to_selections()),
|
||||
multi_agent_version: turn.multi_agent_version,
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
@@ -60,6 +60,7 @@ use codex_mcp::ToolInfo;
|
||||
use codex_protocol::dynamic_tools::DynamicToolSpec;
|
||||
use codex_protocol::openai_models::ConfigShellToolType;
|
||||
use codex_protocol::openai_models::InputModality;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::openai_models::ToolMode;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
@@ -285,14 +286,6 @@ fn namespace_tools_enabled(turn_context: &TurnContext) -> bool {
|
||||
turn_context.provider.capabilities().namespace_tools
|
||||
}
|
||||
|
||||
fn multi_agent_v2_enabled(turn_context: &TurnContext) -> bool {
|
||||
turn_context.features.get().enabled(Feature::MultiAgentV2)
|
||||
}
|
||||
|
||||
fn collab_tools_enabled(turn_context: &TurnContext) -> bool {
|
||||
multi_agent_v2_enabled(turn_context) || turn_context.features.get().enabled(Feature::Collab)
|
||||
}
|
||||
|
||||
fn goal_tools_enabled(turn_context: &TurnContext) -> bool {
|
||||
turn_context.goal_tools_enabled()
|
||||
&& !matches!(
|
||||
@@ -351,7 +344,7 @@ fn standalone_image_generation_available(
|
||||
}
|
||||
|
||||
fn wait_agent_timeout_options(turn_context: &TurnContext) -> WaitAgentTimeoutOptions {
|
||||
if multi_agent_v2_enabled(turn_context) {
|
||||
if turn_context.multi_agent_version == Some(MultiAgentVersion::V2) {
|
||||
return WaitAgentTimeoutOptions {
|
||||
default_timeout_ms: turn_context.config.multi_agent_v2.default_wait_timeout_ms,
|
||||
min_timeout_ms: turn_context.config.multi_agent_v2.min_wait_timeout_ms,
|
||||
@@ -367,7 +360,7 @@ fn wait_agent_timeout_options(turn_context: &TurnContext) -> WaitAgentTimeoutOpt
|
||||
}
|
||||
|
||||
fn max_concurrent_threads_per_session(turn_context: &TurnContext) -> Option<usize> {
|
||||
multi_agent_v2_enabled(turn_context).then_some(
|
||||
(turn_context.multi_agent_version == Some(MultiAgentVersion::V2)).then_some(
|
||||
turn_context
|
||||
.config
|
||||
.multi_agent_v2
|
||||
@@ -646,8 +639,16 @@ fn add_core_utility_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut
|
||||
|
||||
fn add_collaboration_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) {
|
||||
let turn_context = context.turn_context;
|
||||
if collab_tools_enabled(turn_context) {
|
||||
if multi_agent_v2_enabled(turn_context) {
|
||||
if matches!(
|
||||
&turn_context.session_source,
|
||||
SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. })
|
||||
if turn_context.multi_agent_version == Some(MultiAgentVersion::V1)
|
||||
&& *depth >= turn_context.config.agent_max_depth
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if turn_context.multi_agent_version.is_some() {
|
||||
if turn_context.multi_agent_version == Some(MultiAgentVersion::V2) {
|
||||
let exposure = if turn_context.config.multi_agent_v2.non_code_mode_only {
|
||||
ToolExposure::DirectModelOnly
|
||||
} else {
|
||||
|
||||
@@ -8,11 +8,13 @@ use codex_mcp::ToolInfo;
|
||||
use codex_model_provider::create_model_provider;
|
||||
use codex_model_provider_info::AMAZON_BEDROCK_PROVIDER_ID;
|
||||
use codex_model_provider_info::ModelProviderInfo;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::config_types::WebSearchMode;
|
||||
use codex_protocol::dynamic_tools::DynamicToolSpec;
|
||||
use codex_protocol::openai_models::ApplyPatchToolType;
|
||||
use codex_protocol::openai_models::ConfigShellToolType;
|
||||
use codex_protocol::openai_models::InputModality;
|
||||
use codex_protocol::openai_models::MultiAgentVersion;
|
||||
use codex_protocol::openai_models::ToolMode;
|
||||
use codex_protocol::openai_models::WebSearchToolType;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
@@ -30,8 +32,10 @@ use codex_tools::ToolSpec;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
|
||||
use crate::guardian::GUARDIAN_REVIEWER_NAME;
|
||||
use crate::session::tests::make_session_and_context;
|
||||
use crate::session::turn_context::TurnContext;
|
||||
use crate::session::turn_context::resolve_multi_agent_version;
|
||||
use crate::tools::handlers::multi_agents_spec::MULTI_AGENT_V1_NAMESPACE;
|
||||
use crate::tools::router::ToolRouter;
|
||||
use crate::tools::router::ToolRouterParams;
|
||||
@@ -225,6 +229,15 @@ fn set_feature(turn: &mut TurnContext, feature: Feature, enabled: bool) {
|
||||
ToolMode::Direct
|
||||
}
|
||||
});
|
||||
turn.multi_agent_version = {
|
||||
if turn.config.features.enabled(Feature::MultiAgentV2) {
|
||||
Some(MultiAgentVersion::V2)
|
||||
} else if turn.config.features.enabled(Feature::Collab) {
|
||||
Some(MultiAgentVersion::V1)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
fn set_features(turn: &mut TurnContext, features: &[Feature]) {
|
||||
@@ -821,6 +834,109 @@ async fn tool_mode_selector_overrides_feature_flags() {
|
||||
]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolved_multi_agent_version_controls_runtime_behavior() {
|
||||
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_input",
|
||||
"resume_agent",
|
||||
"wait_agent",
|
||||
"close_agent",
|
||||
"send_message",
|
||||
"followup_task",
|
||||
"list_agents",
|
||||
]);
|
||||
assert_eq!(
|
||||
v1.namespace_function_names(MULTI_AGENT_V1_NAMESPACE),
|
||||
&[
|
||||
"close_agent".to_string(),
|
||||
"resume_agent".to_string(),
|
||||
"send_input".to_string(),
|
||||
"spawn_agent".to_string(),
|
||||
"wait_agent".to_string(),
|
||||
]
|
||||
);
|
||||
|
||||
let v2 = probe(|turn| {
|
||||
set_feature(turn, Feature::Collab, /*enabled*/ false);
|
||||
set_feature(turn, Feature::MultiAgentV2, /*enabled*/ false);
|
||||
turn.multi_agent_version = Some(MultiAgentVersion::V2);
|
||||
})
|
||||
.await;
|
||||
|
||||
v2.assert_visible_contains(&[
|
||||
"spawn_agent",
|
||||
"send_message",
|
||||
"followup_task",
|
||||
"wait_agent",
|
||||
"close_agent",
|
||||
"list_agents",
|
||||
]);
|
||||
|
||||
let review = probe(|turn| {
|
||||
turn.session_source = SessionSource::SubAgent(SubAgentSource::Review);
|
||||
turn.multi_agent_version = resolve_multi_agent_version(
|
||||
turn.config.as_ref(),
|
||||
&turn.session_source,
|
||||
/*inherited_multi_agent_version*/ None,
|
||||
);
|
||||
})
|
||||
.await;
|
||||
review.assert_visible_lacks(&[
|
||||
MULTI_AGENT_V1_NAMESPACE,
|
||||
"spawn_agent",
|
||||
"send_message",
|
||||
"followup_task",
|
||||
"wait_agent",
|
||||
"close_agent",
|
||||
"list_agents",
|
||||
]);
|
||||
|
||||
let guardian_review = probe(|turn| {
|
||||
turn.session_source =
|
||||
SessionSource::SubAgent(SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()));
|
||||
turn.multi_agent_version = resolve_multi_agent_version(
|
||||
turn.config.as_ref(),
|
||||
&turn.session_source,
|
||||
/*inherited_multi_agent_version*/ None,
|
||||
);
|
||||
})
|
||||
.await;
|
||||
guardian_review.assert_visible_lacks(&[
|
||||
MULTI_AGENT_V1_NAMESPACE,
|
||||
"spawn_agent",
|
||||
"send_message",
|
||||
"followup_task",
|
||||
"wait_agent",
|
||||
"close_agent",
|
||||
"list_agents",
|
||||
]);
|
||||
|
||||
let max_depth_v1 = probe(|turn| {
|
||||
turn.session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id: ThreadId::new(),
|
||||
depth: turn.config.agent_max_depth,
|
||||
agent_path: None,
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
});
|
||||
turn.multi_agent_version = resolve_multi_agent_version(
|
||||
turn.config.as_ref(),
|
||||
&turn.session_source,
|
||||
Some(MultiAgentVersion::V1),
|
||||
);
|
||||
})
|
||||
.await;
|
||||
max_depth_v1.assert_visible_lacks(&[MULTI_AGENT_V1_NAMESPACE]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn v1_multi_agent_tools_defer_when_tool_search_available() {
|
||||
let plan = probe(|turn| {
|
||||
|
||||
Reference in New Issue
Block a user