mirror of
https://github.com/openai/codex.git
synced 2026-10-27 18:48:37 +00:00
Sync subagent multi-agent mode with root
This commit is contained in:
@@ -17,6 +17,7 @@ use crate::thread_rollout_truncation::truncate_rollout_to_last_n_fork_turns;
|
||||
use codex_protocol::AgentPath;
|
||||
use codex_protocol::SessionId;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::config_types::MultiAgentMode;
|
||||
use codex_protocol::error::CodexErr;
|
||||
use codex_protocol::error::Result as CodexResult;
|
||||
use codex_protocol::models::ContentItem;
|
||||
@@ -121,6 +122,12 @@ impl AgentControl {
|
||||
self.session_id
|
||||
}
|
||||
|
||||
pub(crate) async fn root_multi_agent_mode(&self) -> CodexResult<Option<MultiAgentMode>> {
|
||||
let state = self.upgrade()?;
|
||||
let root_thread = state.get_thread(ThreadId::from(self.session_id)).await?;
|
||||
Ok(root_thread.config_snapshot().await.multi_agent_mode)
|
||||
}
|
||||
|
||||
/// Send rich user input items to an existing agent thread.
|
||||
pub(crate) async fn send_input(
|
||||
&self,
|
||||
|
||||
@@ -830,9 +830,9 @@ async fn spawn_thread_subagent_inherits_parent_selected_multi_agent_mode_without
|
||||
})
|
||||
.await
|
||||
.expect("update parent multi-agent mode");
|
||||
let control = parent_thread.codex.session.services.agent_control.clone();
|
||||
|
||||
let child_thread_id = harness
|
||||
.control
|
||||
let child_thread_id = control
|
||||
.spawn_agent(
|
||||
harness.config.clone(),
|
||||
text_input("child task"),
|
||||
@@ -858,6 +858,27 @@ async fn spawn_thread_subagent_inherits_parent_selected_multi_agent_mode_without
|
||||
child_snapshot.multi_agent_mode,
|
||||
Some(MultiAgentMode::Proactive)
|
||||
);
|
||||
|
||||
parent_thread
|
||||
.codex
|
||||
.session
|
||||
.update_settings(crate::session::SessionSettingsUpdate {
|
||||
multi_agent_mode: Some(MultiAgentMode::ExplicitRequestOnly),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.expect("update parent multi-agent mode after child spawn");
|
||||
let child_thread = harness
|
||||
.manager
|
||||
.get_thread(child_thread_id)
|
||||
.await
|
||||
.expect("child thread should remain registered");
|
||||
let child_turn = child_thread.codex.session.new_default_turn().await;
|
||||
|
||||
assert_eq!(
|
||||
child_turn.multi_agent_mode,
|
||||
Some(MultiAgentMode::ExplicitRequestOnly)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -2254,7 +2275,7 @@ async fn spawn_thread_subagent_uses_role_specific_nickname_candidates() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resume_thread_subagent_restores_stored_metadata_and_effective_multi_agent_mode() {
|
||||
async fn resumed_thread_subagent_uses_root_multi_agent_mode_for_its_next_turn() {
|
||||
let (home, mut config) = test_config().await;
|
||||
config
|
||||
.features
|
||||
@@ -2277,11 +2298,11 @@ async fn resume_thread_subagent_restores_stored_metadata_and_effective_multi_age
|
||||
control,
|
||||
};
|
||||
let (parent_thread_id, parent_thread) = harness.start_thread().await;
|
||||
let control = parent_thread.codex.session.services.agent_control.clone();
|
||||
let agent_path = AgentPath::from_string("/root/explorer".to_string())
|
||||
.expect("test agent path should be valid");
|
||||
|
||||
let child_thread_id = harness
|
||||
.control
|
||||
let child_thread_id = control
|
||||
.spawn_agent(
|
||||
harness.config.clone(),
|
||||
text_input("hello child"),
|
||||
@@ -2333,8 +2354,7 @@ async fn resume_thread_subagent_restores_stored_metadata_and_effective_multi_age
|
||||
})
|
||||
.await
|
||||
.expect("change parent multi-agent mode before child resume");
|
||||
let mut status_rx = harness
|
||||
.control
|
||||
let mut status_rx = control
|
||||
.subscribe_status(child_thread_id)
|
||||
.await
|
||||
.expect("status subscription should succeed");
|
||||
@@ -2375,14 +2395,12 @@ async fn resume_thread_subagent_restores_stored_metadata_and_effective_multi_age
|
||||
.await
|
||||
.expect("child thread metadata should be persisted to sqlite before shutdown");
|
||||
|
||||
let _ = harness
|
||||
.control
|
||||
let _ = control
|
||||
.shutdown_live_agent(child_thread_id)
|
||||
.await
|
||||
.expect("child shutdown should submit");
|
||||
|
||||
let resumed_thread_id = harness
|
||||
.control
|
||||
let resumed_thread_id = control
|
||||
.resume_agent_from_rollout(
|
||||
harness.config.clone(),
|
||||
child_thread_id,
|
||||
@@ -2425,9 +2443,18 @@ async fn resume_thread_subagent_restores_stored_metadata_and_effective_multi_age
|
||||
resumed_snapshot.multi_agent_mode,
|
||||
Some(MultiAgentMode::Proactive)
|
||||
);
|
||||
let resumed_thread = harness
|
||||
.manager
|
||||
.get_thread(resumed_thread_id)
|
||||
.await
|
||||
.expect("resumed child thread should remain registered");
|
||||
let resumed_turn = resumed_thread.codex.session.new_default_turn().await;
|
||||
assert_eq!(
|
||||
resumed_turn.multi_agent_mode,
|
||||
Some(MultiAgentMode::ExplicitRequestOnly)
|
||||
);
|
||||
|
||||
let _ = harness
|
||||
.control
|
||||
let _ = control
|
||||
.shutdown_live_agent(resumed_thread_id)
|
||||
.await
|
||||
.expect("resumed child shutdown should submit");
|
||||
|
||||
@@ -746,10 +746,20 @@ impl Session {
|
||||
async fn new_turn_context_from_configuration(
|
||||
&self,
|
||||
sub_id: String,
|
||||
session_configuration: SessionConfiguration,
|
||||
mut session_configuration: SessionConfiguration,
|
||||
final_output_json_schema: Option<Option<Value>>,
|
||||
multi_agent_runtime: TurnMultiAgentRuntime,
|
||||
) -> Arc<TurnContext> {
|
||||
// Thread-spawn subagents follow the root's current selection. If the root is unavailable,
|
||||
// retain the subagent's persisted selection so standalone resume still has a fallback.
|
||||
if matches!(
|
||||
session_configuration.session_source,
|
||||
SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. })
|
||||
) && let Ok(root_multi_agent_mode) =
|
||||
self.services.agent_control.root_multi_agent_mode().await
|
||||
{
|
||||
session_configuration.multi_agent_mode = root_multi_agent_mode;
|
||||
}
|
||||
let turn_environments = self.services.turn_environments.snapshot().await;
|
||||
let primary_turn_environment = turn_environments.primary().cloned();
|
||||
// TODO(anp): Migrate per-turn config and legacy TurnContext cwd consumers to PathUri so
|
||||
|
||||
@@ -7,8 +7,10 @@ use codex_protocol::protocol::Op;
|
||||
use codex_protocol::protocol::ThreadSettingsOverrides;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
use core_test_support::responses::ev_completed;
|
||||
use core_test_support::responses::ev_function_call;
|
||||
use core_test_support::responses::ev_response_created;
|
||||
use core_test_support::responses::mount_sse_once;
|
||||
use core_test_support::responses::mount_sse_once_match;
|
||||
use core_test_support::responses::mount_sse_sequence;
|
||||
use core_test_support::responses::sse;
|
||||
use core_test_support::responses::start_mock_server;
|
||||
@@ -17,6 +19,10 @@ use core_test_support::test_codex::test_codex;
|
||||
use core_test_support::wait_for_event;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
use std::time::Duration;
|
||||
|
||||
const SPAWN_CALL_ID: &str = "spawn-call-1";
|
||||
|
||||
const NO_SPAWN_TEXT: &str = "Do not spawn sub-agents unless the user explicitly asks for sub-agents, delegation, or parallel agent work.";
|
||||
const PROACTIVE_TEXT: &str = "Proactive multi-agent delegation is active.";
|
||||
@@ -59,6 +65,10 @@ async fn submit_turn(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn body_contains(req: &wiremock::Request, text: &str) -> bool {
|
||||
String::from_utf8_lossy(&req.body).contains(text)
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn multi_agent_mode_is_sticky_and_emits_only_on_change() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
@@ -132,6 +142,118 @@ async fn multi_agent_mode_is_sticky_and_emits_only_on_change() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn existing_subagent_uses_root_multi_agent_mode_on_its_next_turn() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let server = start_mock_server().await;
|
||||
let root_prompt = "spawn a child";
|
||||
let child_prompt = "child initial task";
|
||||
let child_followup = "child followup task";
|
||||
let spawn_args = serde_json::to_string(&json!({
|
||||
"message": child_prompt,
|
||||
"task_name": "worker",
|
||||
}))?;
|
||||
mount_sse_once_match(
|
||||
&server,
|
||||
move |req: &wiremock::Request| body_contains(req, root_prompt),
|
||||
sse(vec![
|
||||
ev_response_created("root-spawn-response"),
|
||||
ev_function_call(SPAWN_CALL_ID, "spawn_agent", &spawn_args),
|
||||
ev_completed("root-spawn-response"),
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
mount_sse_once_match(
|
||||
&server,
|
||||
move |req: &wiremock::Request| {
|
||||
body_contains(req, child_prompt) && !body_contains(req, SPAWN_CALL_ID)
|
||||
},
|
||||
sse(vec![
|
||||
ev_response_created("child-initial-response"),
|
||||
ev_completed("child-initial-response"),
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
mount_sse_once_match(
|
||||
&server,
|
||||
|req: &wiremock::Request| body_contains(req, SPAWN_CALL_ID),
|
||||
sse(vec![
|
||||
ev_response_created("root-followup-response"),
|
||||
ev_completed("root-followup-response"),
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
let child_followup_response = mount_sse_once_match(
|
||||
&server,
|
||||
move |req: &wiremock::Request| body_contains(req, child_followup),
|
||||
sse(vec![
|
||||
ev_response_created("child-followup-response"),
|
||||
ev_completed("child-followup-response"),
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
|
||||
let test = test_codex()
|
||||
.with_config(|config| {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::Collab)
|
||||
.expect("test config should allow feature update");
|
||||
config
|
||||
.features
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("test config should allow feature update");
|
||||
config
|
||||
.features
|
||||
.enable(Feature::MultiAgentMode)
|
||||
.expect("test config should allow feature update");
|
||||
config
|
||||
.features
|
||||
.disable(Feature::EnableRequestCompression)
|
||||
.expect("test config should allow feature update");
|
||||
})
|
||||
.build(&server)
|
||||
.await?;
|
||||
let mut created_threads = test.thread_manager.subscribe_thread_created();
|
||||
|
||||
submit_turn(&test.codex, root_prompt, /*mode*/ None).await?;
|
||||
let child_thread_id =
|
||||
tokio::time::timeout(Duration::from_secs(10), created_threads.recv()).await??;
|
||||
let child_thread = test.thread_manager.get_thread(child_thread_id).await?;
|
||||
wait_for_event(&child_thread, |event| {
|
||||
matches!(event, EventMsg::TurnComplete(_))
|
||||
})
|
||||
.await;
|
||||
test.codex
|
||||
.submit(Op::ThreadSettings {
|
||||
thread_settings: ThreadSettingsOverrides {
|
||||
multi_agent_mode: Some(MultiAgentMode::Proactive),
|
||||
..Default::default()
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
wait_for_event(&test.codex, |event| {
|
||||
matches!(event, EventMsg::ThreadSettingsApplied(_))
|
||||
})
|
||||
.await;
|
||||
|
||||
submit_turn(&child_thread, child_followup, /*mode*/ None).await?;
|
||||
|
||||
let input = child_followup_response.single_request().input();
|
||||
let texts = developer_texts(&input);
|
||||
assert_eq!(
|
||||
(
|
||||
count_containing(&texts, MULTI_AGENT_MODE_OPEN_TAG),
|
||||
count_containing(&texts, NO_SPAWN_TEXT),
|
||||
count_containing(&texts, PROACTIVE_TEXT),
|
||||
),
|
||||
(2, 1, 1)
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn multi_agent_mode_feature_uses_explicit_mode_when_disabled() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
Reference in New Issue
Block a user