From c1934fc99e6b40ddff4261c910ce7733e3abea26 Mon Sep 17 00:00:00 2001 From: sayan-oai Date: Wed, 11 Nov 2026 09:59:20 +0000 Subject: [PATCH] Resolve saved Daybreak choices for each subagent turn (#60773) ## Why A subagent can use a different model from its parent, with different supported access programs. Saved Daybreak choices need to resolve against the subagent's model and remain consistent throughout its turn, including compaction. ## What changed - Resolve saved choices before native subagent turns start, preserving explicit turn overrides and applying saved toggles on later turns. - Distinguish inherited parent programs from explicit child selections. Adapt inherited Daybreak Blue/Red selections to the child model while retaining the saved choice as the default for later turns and nested subagents. - Validate the account associated with automatic selections and reject incompatible saved choices before sending model requests, while retaining rejected user input in conversation history. - Require model changes to wait until the next turn when a saved Daybreak choice is present. ## Testing Add coverage for both multi-agent spawn versions, model-specific selection, automatic compaction, WebSocket response reuse, account changes, unsupported choices, and rejected-input persistence across resume. GitOrigin-RevId: 11971138d656600d9878187225e4d33217e68ad4 --- codex-rs/core/src/agent/control/spawn.rs | 120 +++- codex-rs/core/src/agent/helper_daybreak.rs | 1 - .../core/src/agent/helper_daybreak/turn.rs | 121 +++- codex-rs/core/src/agent/types.rs | 10 + codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/session/step_activation.rs | 14 + codex-rs/core/src/session/turn_context.rs | 2 +- codex-rs/core/src/tasks/regular.rs | 14 + .../thread_manager/helper_daybreak_tests.rs | 230 +++++-- codex-rs/core/src/thread_spawn.rs | 5 +- .../src/tools/handlers/multi_agents/spawn.rs | 5 +- .../core/tests/suite/cyber_access_program.rs | 602 ++++++++++++++++++ 12 files changed, 1043 insertions(+), 82 deletions(-) diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index c026595f61..c7e97c62b4 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -12,6 +12,7 @@ use crate::agent::role::apply_role_to_config; use crate::agent::types::AgentMetadata; use crate::agent::types::LiveAgent; use crate::agent::types::SpawnAgentOptions; +use crate::agent::types::SpawnProgramSource; use crate::agents_md_manager::SessionInstructions; use crate::codex_thread::CodexThread; use crate::codex_thread::ThreadConfigSnapshot; @@ -746,51 +747,86 @@ impl LocalAgentControl { disabled_plugin_ids, ..StartThreadOptions::new(config.clone()) }; + let mut cyber_access_program = options.cyber_access_program; + let mut inherited_program_to_resolve = None; if let Some(parent_id) = session_source .as_ref() .and_then(SessionSource::parent_thread_id) && let Ok(parent) = state.get_thread(parent_id).await - && config.features.enabled(Feature::DaybreakProgramPropagation) - && let Some(enrollment) = parent - .thread_extension_data() - .get::() - && parent - .session - .get_config() - .await - .features - .enabled(Feature::DaybreakProgramPropagation) { - start_options - .thread_extension_init - .insert(helper_daybreak::Enrollment::NativeHelper); + let turn = match options.parent_turn_id.as_deref() { + Some(id) => parent.session.turn_context_for_sub_id(id).await, + None => None, + }; let spawn_owner = helper_daybreak::eligible_owner(&parent.session); - if let Some(turn_id) = options.parent_turn_id.as_deref() - && let Some(turn) = parent.session.turn_context_for_sub_id(turn_id).await - && let Some(intent) = helper_daybreak::spawning_intent( - &parent.session, - &turn, - &enrollment, - spawn_owner.as_ref(), - ) - .await + if options.cyber_access_program_source == SpawnProgramSource::ParentTurn + && let Some(resolution) = turn.as_ref().and_then(|turn| { + turn.extension_data + .get::() + }) { - if let Err(error) = &intent.choice { - // Only a usable, successful read can become a child's inherited default. - if options.cyber_access_program.is_none() - && config.model_provider_id == codex_model_provider_info::OPENAI_PROVIDER_ID + resolution + .validate() + .map_err(|err| err.with_agent_context(AgentErrorContext::ChildStartup))?; + // Carry the saved choice. The child resolves it for its own model and account. + cyber_access_program = None; + } + if config.features.enabled(Feature::DaybreakProgramPropagation) + && let Some(enrollment) = parent + .thread_extension_data() + .get::() + && parent + .session + .get_config() + .await + .features + .enabled(Feature::DaybreakProgramPropagation) + { + start_options + .thread_extension_init + .insert(helper_daybreak::Enrollment::NativeHelper); + if let Some(turn) = turn { + let propagate_program = spawn_owner.is_some() + && turn.config.model_provider_id + == codex_model_provider_info::OPENAI_PROVIDER_ID + && codex_model_provider::provider_uses_first_party_auth_path( + turn.provider.info(), + ) + && config.model_provider_id + == codex_model_provider_info::OPENAI_PROVIDER_ID && config .model_provider .supports_cyber_access_programs(&config.model_provider_id) && codex_model_provider::provider_uses_first_party_auth_path( &config.model_provider, - ) + ); + if propagate_program + && options.cyber_access_program_source == SpawnProgramSource::ParentTurn { - return Err(CodexErr::InvalidRequest(error.clone()) - .with_agent_context(AgentErrorContext::ChildStartup)); + inherited_program_to_resolve = cyber_access_program; + } + if let Some(intent) = helper_daybreak::spawning_intent( + &parent.session, + &turn, + &enrollment, + spawn_owner.as_ref(), + ) + .await + { + // A read failure must not become an immutable default on a live child. + if propagate_program + && (options.cyber_access_program.is_none() + || options.cyber_access_program_source + == SpawnProgramSource::ParentTurn) + && let Err(error) = &intent.choice + { + return Err(CodexErr::InvalidRequest(error.clone()) + .with_agent_context(AgentErrorContext::ChildStartup)); + } + if intent.choice.is_ok() { + start_options.thread_extension_init.insert(intent); + } } - } else { - start_options.thread_extension_init.insert(intent); } } } @@ -898,11 +934,31 @@ impl LocalAgentControl { None }; + if let Some(inherited) = inherited_program_to_resolve + && inherited != codex_protocol::turn_input::CyberAccessProgram::Standard + && let Some(model) = new_thread + .thread + .thread_extension_data() + .get::() + && model + .available_access_programs + .as_ref() + .is_some_and(|programs| !programs.cyber.contains(&inherited)) + { + // Preserve the parent's turn override, adapting Blue/Red to the child model. + // The separately captured saved choice remains the default for later turns. + cyber_access_program = helper_daybreak::turn::select_program( + &model, + codex_protocol::openai_models::SavedDaybreakChoice::On, + ) + .map_err(CodexErr::InvalidRequest)? + .or(Some(inherited)); + } let start_options = TurnStartOptions { parent_turn_id: options.parent_turn_id, turn_trigger: options.turn_trigger, root_turn_id: options.root_turn_id, - cyber_access_program: options.cyber_access_program, + cyber_access_program, ..Default::default() }; let input_admission_started_at = Instant::now(); diff --git a/codex-rs/core/src/agent/helper_daybreak.rs b/codex-rs/core/src/agent/helper_daybreak.rs index 430b41d543..890cf5e968 100644 --- a/codex-rs/core/src/agent/helper_daybreak.rs +++ b/codex-rs/core/src/agent/helper_daybreak.rs @@ -54,7 +54,6 @@ pub(crate) async fn spawning_intent( let choice = match saved_choice(parent).await { Ok(Some(enabled)) => Ok(SavedDaybreakChoice::from(Some(enabled))), Ok(None) => return None, - Err(_) if turn.cyber_access_program.is_some() => return None, Err(error) => Err(error), }; Some(SavedIntent { diff --git a/codex-rs/core/src/agent/helper_daybreak/turn.rs b/codex-rs/core/src/agent/helper_daybreak/turn.rs index 8f899da7b9..5523f76d40 100644 --- a/codex-rs/core/src/agent/helper_daybreak/turn.rs +++ b/codex-rs/core/src/agent/helper_daybreak/turn.rs @@ -1,4 +1,5 @@ -//! Capture the saved Daybreak choice for an admitted native helper turn. +//! Resolve a native helper's saved Daybreak choice once, before its turn starts. +//! The existing turn program then travels with normal and compaction requests. use super::Enrollment; use super::SavedIntent; use super::eligible_owner; @@ -7,7 +8,40 @@ use crate::session::session::Session; use crate::session::turn_context::TurnContext; use codex_features::Feature; use codex_login::AuthOwner; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; +use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::SavedDaybreakChoice; +use codex_protocol::turn_input::CyberAccessProgram; + +const ACCOUNT_CHANGED: &str = + "The account for this subagent's saved Daybreak choice has changed. Start a new subagent."; + +/// An implicit program stays bound to the account whose choice selected it. +pub(crate) struct SavedProgramResolution(Result); + +impl SavedProgramResolution { + pub(crate) fn validate(&self) -> CodexResult<()> { + match &self.0 { + Ok(owner) if owner.is_current() => Ok(()), + Ok(_) => Err(CodexErr::InvalidRequest(ACCOUNT_CHANGED.to_owned())), + Err(error) => Err(CodexErr::InvalidRequest(error.clone())), + } + } +} + +/// A helper resolves automatic choices from saved intent on its own turn. +pub(crate) fn explicit_program_for_spawn(turn: &TurnContext) -> Option { + if turn + .extension_data + .get::() + .is_some() + { + None + } else { + turn.cyber_access_program + } +} /// The chat account at admission, captured before preparing the new turn. pub(crate) struct HelperTurnAdmission(pub(crate) AuthOwner); @@ -18,16 +52,28 @@ impl HelperTurnAdmission { if *enrollment != Enrollment::NativeHelper { return None; } - eligible_owner(session).map(Self) + eligible_owner(session) + .or_else(|| { + let inherited = session + .services + .thread_extension_data + .get::()?; + (!inherited.owner.is_current()).then(|| inherited.owner.clone()) + }) + .map(Self) } - pub(crate) async fn capture_intent(&self, session: &Session, turn: &TurnContext) { + pub(crate) async fn capture_intent(&self, session: &Session, turn: &mut TurnContext) { if !turn .config .features .enabled(Feature::DaybreakProgramPropagation) || turn.config.model_provider_id != codex_model_provider_info::OPENAI_PROVIDER_ID || !codex_model_provider::provider_uses_first_party_auth_path(turn.provider.info()) + || !turn + .config + .model_provider + .supports_cyber_access_programs(&turn.config.model_provider_id) { return; } @@ -43,6 +89,75 @@ impl HelperTurnAdmission { Err(_) if turn.cyber_access_program.is_some() => return, Err(error) => (Err(error), self.0.clone()), }; + if turn.cyber_access_program.is_none() { + let resolution = if owner != self.0 || eligible_owner(session).as_ref() != Some(&self.0) + { + Err(ACCOUNT_CHANGED.to_owned()) + } else { + choice + .clone() + .and_then(|choice| select_program(turn.model_info(), choice)) + }; + let bound_owner = resolution.map(|program| { + turn.cyber_access_program = program; + owner.clone() + }); + turn.extension_data + .insert(SavedProgramResolution(bound_owner)); + } turn.extension_data.insert(SavedIntent { choice, owner }); } } + +pub(crate) fn select_program( + model: &ModelInfo, + choice: SavedDaybreakChoice, +) -> Result, String> { + let Some(programs) = model.available_access_programs.as_ref() else { + return Ok(None); + }; + let selected = programs.select_saved(choice); + if selected.is_none() { + match choice { + SavedDaybreakChoice::On if !programs.cyber.is_empty() => { + return Err(format!( + "Model {} does not advertise Daybreak support. Turn Daybreak off or choose a compatible model.", + model.slug + )); + } + SavedDaybreakChoice::Off => { + return Err(format!( + "Model {} does not advertise Standard. Choose a model that supports turning Daybreak off.", + model.slug + )); + } + SavedDaybreakChoice::On | SavedDaybreakChoice::Unset => {} + } + } + Ok(selected) +} + +#[cfg(test)] +mod tests { + use super::*; + use codex_protocol::openai_models::ModelAccessPrograms; + + #[test] + fn missing_metadata_and_unset_choice_preserve_implicit_selection() { + let mut model = codex_models_manager::bundled_models_response() + .unwrap() + .models + .remove(0); + model.available_access_programs = None; + assert_eq!(select_program(&model, SavedDaybreakChoice::Off), Ok(None)); + for cyber in [vec![], vec![CyberAccessProgram::DaybreakRed]] { + model.available_access_programs = Some(ModelAccessPrograms { cyber }); + assert_eq!(select_program(&model, SavedDaybreakChoice::Unset), Ok(None)); + assert!( + select_program(&model, SavedDaybreakChoice::Off) + .unwrap_err() + .contains("does not advertise Standard") + ); + } + } +} diff --git a/codex-rs/core/src/agent/types.rs b/codex-rs/core/src/agent/types.rs index 2b3626fd51..55699aae2e 100644 --- a/codex-rs/core/src/agent/types.rs +++ b/codex-rs/core/src/agent/types.rs @@ -23,6 +23,14 @@ pub enum SpawnAgentForkMode { FullHistory, } +/// Whether a spawn program was chosen for the child or copied from its parent turn. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum SpawnProgramSource { + #[default] + Explicit, + ParentTurn, +} + #[derive(Clone, Debug, Default)] pub struct SpawnAgentOptions { pub fork_parent_spawn_call_id: Option, @@ -39,6 +47,8 @@ pub struct SpawnAgentOptions { pub disabled_plugin_ids: Option>, pub multi_agent_v2_usage_hints: Option, pub cyber_access_program: Option, + /// Parent selections apply to the first turn and may need adapting to the child's model. + pub cyber_access_program_source: SpawnProgramSource, } /// Identity and status observed from a loaded agent, without a handle to its runtime. diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index d684b39426..90d089405b 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -81,6 +81,7 @@ pub use agent::types::MessageDeliveryMode; pub use agent::types::ResolvedMultiAgentV2UsageHints; pub use agent::types::SpawnAgentForkMode; pub use agent::types::SpawnAgentOptions; +pub use agent::types::SpawnProgramSource; pub use rollout_budget::RolloutBudgetReminder; mod agent_communication; pub use agent_communication::PendingAgentMessage; diff --git a/codex-rs/core/src/session/step_activation.rs b/codex-rs/core/src/session/step_activation.rs index d89a1f8362..ff09dd5f88 100644 --- a/codex-rs/core/src/session/step_activation.rs +++ b/codex-rs/core/src/session/step_activation.rs @@ -269,6 +269,20 @@ impl Session { let Some((turn_context, task_done, current)) = target else { return TurnSettingsUpdateOutcome::TargetUnavailable; }; + if update + .model + .as_ref() + .is_some_and(|model| *model != current.model_info.slug) + && turn_context + .extension_data + .get::() + .is_some() + { + return TurnSettingsUpdateOutcome::Rejected { + reason: "Change the model on the next turn when using a saved Daybreak choice" + .to_string(), + }; + } let TurnSettingsUpdate { approvals_reviewer, environments: environment_requests, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 2fe58d9dab..794fa420c0 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -1491,7 +1491,7 @@ impl Session { turn_context.cyber_access_program = cyber_access_program::for_provider(&turn_context.config, options.cyber_access_program); if let Some(admission) = options.helper_admission { - admission.capture_intent(self, &turn_context).await; + admission.capture_intent(self, &mut turn_context).await; } // Tool observers retain this turn's selection even after a later turn starts. if let Some(program) = turn_context.cyber_access_program { diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index 95bcb95102..1923d1fbe8 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -1,4 +1,5 @@ use crate::McpRefreshReason; +use crate::agent::helper_daybreak::turn::SavedProgramResolution; use std::sync::Arc; use codex_async_utils::OrCancelExt; @@ -128,6 +129,19 @@ impl SessionTask for RegularTask { Some(*prewarmed_client_session) } }; + if let Some(resolution) = ctx.extension_data.get::() + && let Err(error) = resolution.validate() + { + let _ = run_hooks_and_record_inputs( + &sess, + &ctx, + &ctx.capture_current_model_info(), + &input, + PersistContext::Standard, + ) + .await; + return Err(error); + } Self::run_turns( sess, ctx, diff --git a/codex-rs/core/src/thread_manager/helper_daybreak_tests.rs b/codex-rs/core/src/thread_manager/helper_daybreak_tests.rs index f754bf63a5..d342e74527 100644 --- a/codex-rs/core/src/thread_manager/helper_daybreak_tests.rs +++ b/codex-rs/core/src/thread_manager/helper_daybreak_tests.rs @@ -8,6 +8,7 @@ use crate::agent::helper_daybreak::saved_choice; use crate::agent::helper_daybreak::turn::HelperTurnAdmission; use crate::agent::next_thread_spawn_depth; use crate::agent::types::SpawnAgentOptions; +use crate::agent::types::SpawnProgramSource; use crate::config::Config; use crate::config::test_config; use crate::session::turn_context::NewTurnContextOptions; @@ -78,6 +79,7 @@ async fn spawn_child( config: &Config, parent: &CodexThread, spawning_turn: &TurnContext, + program_source: SpawnProgramSource, ) -> codex_protocol::error::Result> { let parent_id = parent.session.thread_id(); let source = thread_spawn_source( @@ -101,6 +103,7 @@ async fn spawn_child( parent_thread_id: Some(parent_id), parent_turn_id: Some(spawning_turn.sub_id.clone()), cyber_access_program: spawning_turn.cyber_access_program, + cyber_access_program_source: program_source, ..Default::default() }, ) @@ -133,6 +136,48 @@ fn saved(store: &ExtensionData) -> Option { }) } +async fn paused_helpers( + home: &std::path::Path, + programs: Option>, +) -> (Config, ThreadManager, mpsc::UnboundedReceiver) { + let mut config = test_config().await; + config.codex_home = home.abs(); + config.cwd = home.abs(); + config.agent_max_depth = 2; + config.model_provider.supports_websockets = false; + config.model_provider.base_url = Some("http://127.0.0.1:9/v1".to_owned()); + config.chatgpt_base_url = "http://127.0.0.1:9".to_owned(); + let mut catalog = codex_models_manager::bundled_models_response().expect("local catalog"); + if let Some(programs) = programs { + for model in &mut catalog.models { + model.available_access_programs = + Some(codex_protocol::openai_models::ModelAccessPrograms { + cyber: programs.clone(), + }); + } + } + config.model_catalog = Some(catalog); + config + .features + .enable(Feature::DaybreakProgramPropagation) + .expect("enable propagation"); + let mut manager = ThreadManager::with_models_provider_home_and_state_for_tests( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + config.model_provider.clone(), + config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + crate::init_state_db(&config).await, + ); + let (tx, turns) = mpsc::unbounded_channel(); + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.turn_lifecycle_contributor(Arc::new(PauseAdmittedTurn(tx))); + Arc::get_mut(&mut manager.state) + .expect("unshared manager") + .extensions = Arc::new(extensions.build()); + + (config, manager, turns) +} + #[test_case::test_case(Some(false), None, None; "saved off")] #[test_case::test_case(None, Some(false), None; "pending off before materialization")] #[test_case::test_case(None, None, None; "unset")] @@ -145,33 +190,7 @@ async fn native_spawn_captures_the_saved_choice( explicit: Option, ) { let home = tempdir().expect("create test home"); - let mut config = test_config().await; - config.codex_home = home.path().abs(); - config.cwd = home.path().abs(); - config.agent_max_depth = 2; - config.model_provider.supports_websockets = false; - config.model_provider.base_url = Some("http://127.0.0.1:9/v1".to_owned()); - config.chatgpt_base_url = "http://127.0.0.1:9".to_owned(); - config.model_catalog = - Some(codex_models_manager::bundled_models_response().expect("local catalog")); - config - .features - .enable(Feature::DaybreakProgramPropagation) - .expect("enable propagation"); - let mut manager = ThreadManager::with_models_provider_home_and_state_for_tests( - CodexAuth::create_dummy_chatgpt_auth_for_testing(), - config.model_provider.clone(), - config.codex_home.to_path_buf(), - Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), - crate::init_state_db(&config).await, - ); - let (tx, mut turns) = mpsc::unbounded_channel(); - let mut extensions = ExtensionRegistryBuilder::new(); - extensions.turn_lifecycle_contributor(Arc::new(PauseAdmittedTurn(tx))); - Arc::get_mut(&mut manager.state) - .expect("unshared manager") - .extensions = Arc::new(extensions.build()); - + let (config, manager, mut turns) = paused_helpers(home.path(), /*programs*/ None).await; let mut start_options = StartThreadOptions::new(config.clone()); let thread_store = &manager.state.thread_store; if let Some(enabled) = pending { @@ -221,10 +240,16 @@ async fn native_spawn_captures_the_saved_choice( let backup = path.with_extension("backup"); std::fs::rename(&path, &backup).expect("move healthy rollout"); std::fs::write(&path, b"invalid metadata\n").expect("inject failed metadata read"); - let error = spawn_child(&manager, &config, &root.thread, &root_turn) - .await - .err() - .expect("required read must fail spawn"); + let error = spawn_child( + &manager, + &config, + &root.thread, + &root_turn, + SpawnProgramSource::ParentTurn, + ) + .await + .err() + .expect("required read must fail spawn"); assert!( error .to_string() @@ -235,9 +260,15 @@ async fn native_spawn_captures_the_saved_choice( std::fs::remove_file(&path).expect("remove invalid rollout"); std::fs::rename(backup, path).expect("restore storage before retry"); } - let child = spawn_child(&manager, &config, &root.thread, &root_turn) - .await - .expect("start child"); + let child = spawn_child( + &manager, + &config, + &root.thread, + &root_turn, + SpawnProgramSource::ParentTurn, + ) + .await + .expect("start child"); let child_turn = admitted_turn(&mut turns, &child).await; let expected = stored .or(pending) @@ -262,9 +293,15 @@ async fn native_spawn_captures_the_saved_choice( Some(SavedDaybreakChoice::On) ); save_choice(&root.thread, /*enabled*/ false).await; - let sibling = spawn_child(&manager, &config, &root.thread, &root_turn) - .await - .expect("start sibling after preference change"); + let sibling = spawn_child( + &manager, + &config, + &root.thread, + &root_turn, + SpawnProgramSource::ParentTurn, + ) + .await + .expect("start sibling after preference change"); let sibling_turn = admitted_turn(&mut turns, &sibling).await; assert_eq!( saved(&sibling_turn.extension_data), @@ -288,9 +325,15 @@ async fn native_spawn_captures_the_saved_choice( .await, crate::ConfigRefreshOutcome::Published ); - let unenrolled = spawn_child(&manager, &config, &root.thread, &root_turn) - .await - .expect("start child after parent rollout withdrawal"); + let unenrolled = spawn_child( + &manager, + &config, + &root.thread, + &root_turn, + SpawnProgramSource::ParentTurn, + ) + .await + .expect("start child after parent rollout withdrawal"); let unenrolled_turn = admitted_turn(&mut turns, &unenrolled).await; assert_eq!(saved(&unenrolled_turn.extension_data), None); assert!( @@ -330,9 +373,15 @@ async fn native_spawn_captures_the_saved_choice( // A local toggle affects later turns. This nested spawn belongs to the turn above. save_choice(&child, /*enabled*/ false).await; - let nested = spawn_child(&manager, &config, &child, &explicit_turn) - .await - .expect("start nested child"); + let nested = spawn_child( + &manager, + &config, + &child, + &explicit_turn, + SpawnProgramSource::ParentTurn, + ) + .await + .expect("start nested child"); let nested_turn = admitted_turn(&mut turns, &nested).await; assert_eq!(nested_turn.cyber_access_program, direct); assert_eq!( @@ -370,3 +419,98 @@ async fn native_spawn_captures_the_saved_choice( .await .expect("shut down root"); } + +#[tokio::test] +async fn copied_program_rechecks_the_spawning_turn_owner() { + let home = tempdir().expect("create test home"); + let blue = CyberAccessProgram::DaybreakBlue; + let (config, mut manager, mut turns) = + paused_helpers(home.path(), Some(vec![blue, CyberAccessProgram::Standard])).await; + Arc::get_mut(&mut manager.state) + .expect("unshared manager") + .auth_manager = codex_login::AuthManager::from_auth_for_testing_with_home( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + home.path().to_owned(), + ); + let root = manager + .start_thread(StartThreadOptions::new(config.clone())) + .await + .expect("start root"); + save_choice(&root.thread, /*enabled*/ true).await; + root.thread + .start_or_steer_turn(TurnInputRequest::user_input(text())) + .await + .expect("start root turn"); + let root_turn = admitted_turn(&mut turns, &root.thread).await; + let child = spawn_child( + &manager, + &config, + &root.thread, + &root_turn, + SpawnProgramSource::ParentTurn, + ) + .await + .expect("start child"); + let turn = admitted_turn(&mut turns, &child).await; + assert_eq!(turn.cyber_access_program, Some(blue)); + assert_eq!( + crate::agent::helper_daybreak::turn::explicit_program_for_spawn(&turn), + None + ); + + let auth = manager.auth_manager(); + let previous = auth.auth_owner(); + codex_login::auth::login_with_chatgpt_auth_tokens( + home.path(), + "header.e30.changed", + "other-account", + /*chatgpt_plan_type*/ None, + ) + .expect("switch account"); + auth.reload().await; + assert!(!previous.is_current()); + assert_eq!( + auth.auth_cached() + .and_then(|auth| auth.get_account_id()) + .as_deref(), + Some("other-account") + ); + let err = spawn_child( + &manager, + &config, + &child, + &turn, + SpawnProgramSource::ParentTurn, + ) + .await + .err() + .expect("reject copied program from the previous account"); + assert!( + err.to_string() + .contains("saved Daybreak choice has changed"), + "{err}" + ); + let explicit = spawn_child( + &manager, + &config, + &child, + &turn, + SpawnProgramSource::Explicit, + ) + .await + .expect("accept program explicitly chosen for this child"); + let explicit_turn = admitted_turn(&mut turns, &explicit).await; + assert_eq!(explicit_turn.cyber_access_program, Some(blue)); + assert!( + explicit_turn + .extension_data + .get::() + .is_none() + ); + explicit + .shutdown_and_wait() + .await + .expect("stop explicit child"); + child.shutdown_and_wait().await.expect("stop child"); + root.thread.shutdown_and_wait().await.expect("stop root"); +} diff --git a/codex-rs/core/src/thread_spawn.rs b/codex-rs/core/src/thread_spawn.rs index 4623a50c08..58dab95f05 100644 --- a/codex-rs/core/src/thread_spawn.rs +++ b/codex-rs/core/src/thread_spawn.rs @@ -7,12 +7,14 @@ use crate::agent::api::SpawnRequest; use crate::agent::child_config::SpawnConfigOptions; use crate::agent::child_config::SpawnConfigVersion; use crate::agent::child_config::prepare_agent_spawn_config; +use crate::agent::helper_daybreak::turn::explicit_program_for_spawn; use crate::agent::next_thread_spawn_depth; use crate::agent::types::AgentMessage; use crate::agent::types::LiveAgent; use crate::agent::types::MessageDeliveryMode; use crate::agent::types::SpawnAgentForkMode; use crate::agent::types::SpawnAgentOptions; +use crate::agent::types::SpawnProgramSource; use crate::codex_thread::ThreadConfigSnapshot; use crate::config::Config; use crate::session::multi_agents::resolve_usage_hints; @@ -205,7 +207,8 @@ pub(crate) async fn prepare( environments: Some(step_context.inherited_environments.clone()), disabled_plugin_ids: Some(turn.disabled_plugin_ids.clone()), multi_agent_v2_usage_hints, - cyber_access_program: turn.cyber_access_program, + cyber_access_program: explicit_program_for_spawn(turn), + cyber_access_program_source: SpawnProgramSource::ParentTurn, }, }, task_name, diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index 839874dbbf..42ec6dffe9 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -6,10 +6,12 @@ use crate::agent::child_config::SpawnConfigVersion; use crate::agent::child_config::prepare_agent_spawn_config; use crate::agent::control::render_input_preview; use crate::agent::exceeds_thread_spawn_depth_limit; +use crate::agent::helper_daybreak::turn::explicit_program_for_spawn; use crate::agent::next_thread_spawn_depth; use crate::agent::role::DEFAULT_ROLE_NAME; use crate::agent::types::SpawnAgentForkMode; use crate::agent::types::SpawnAgentOptions; +use crate::agent::types::SpawnProgramSource; use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions; use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v1; use codex_protocol::protocol::MultiAgentVersion; @@ -134,7 +136,8 @@ async fn handle_spawn_agent( environments: Some(step_context.inherited_environments.clone()), disabled_plugin_ids: Some(turn.disabled_plugin_ids.clone()), multi_agent_v2_usage_hints: None, - cyber_access_program: turn.cyber_access_program, + cyber_access_program: explicit_program_for_spawn(turn), + cyber_access_program_source: SpawnProgramSource::ParentTurn, }, }) .await diff --git a/codex-rs/core/tests/suite/cyber_access_program.rs b/codex-rs/core/tests/suite/cyber_access_program.rs index 9a7fbf4645..bea1bb086d 100644 --- a/codex-rs/core/tests/suite/cyber_access_program.rs +++ b/codex-rs/core/tests/suite/cyber_access_program.rs @@ -7,12 +7,20 @@ use codex_core::TurnInputSubmission; use codex_core::TurnStartOptions; use codex_features::Feature; use codex_login::CodexAuth; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::ModelAccessPrograms; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; +use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::protocol::TurnSettingsUpdate; +use codex_protocol::protocol::TurnSettingsUpdateOutcome; use codex_protocol::turn_input::CyberAccessProgram; use codex_protocol::user_input::UserInput; use codex_rollout::RolloutItem; use codex_rollout::RolloutRecorder; +use codex_thread_store::ThreadMetadataPatch; use core_test_support::apps_test_server::AppsTestServer; use core_test_support::apps_test_server::SEARCH_CALENDAR_CREATE_TOOL; use core_test_support::apps_test_server::SEARCH_CALENDAR_NAMESPACE; @@ -23,6 +31,7 @@ use core_test_support::streaming_sse::start_streaming_sse_server; use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use core_test_support::wait_for_successful_turn; use pretty_assertions::assert_eq; use serde_json::json; use test_case::test_case; @@ -631,6 +640,599 @@ async fn cyber_access_program_changes_on_one_websocket_with_response_reuse( Ok(()) } +#[test_case(None; "account changed")] +#[test_case(Some(CyberAccessProgram::DaybreakRed); "saved off without Standard")] +#[test_case(Some(CyberAccessProgram::Standard); "saved on without Daybreak")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn saved_daybreak_rejection_records_child_input( + only_program: Option, +) -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + let requests = responses::mount_response_handler(&server, |request| { + let output = if request.header("x-openai-subagent").is_none() + && request.function_call_output_text("spawn-worker").is_none() + { + responses::ev_function_call_with_namespace( + "spawn-worker", + "collaboration", + "spawn_agent", + &json!({"task_name": "worker", "message": "inspect", "fork_turns": "none"}) + .to_string(), + ) + } else { + responses::ev_assistant_message("done", "Done.") + }; + responses::sse_response(responses::sse(vec![ + output, + responses::ev_completed("done"), + ])) + }) + .await; + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model_info_override("gpt-5.4", move |model| { + model.available_access_programs = Some(ModelAccessPrograms { + cyber: only_program.map_or_else( + || { + vec![ + CyberAccessProgram::DaybreakBlue, + CyberAccessProgram::Standard, + ] + }, + |program| vec![program], + ), + }); + model.use_responses_lite = false; + }) + .with_config(|config| { + for feature in [ + Feature::Collab, + Feature::MultiAgentV2, + Feature::DaybreakProgramPropagation, + ] { + config + .features + .enable(feature) + .expect("enable saved choice test"); + } + }) + .build_with_auto_env(&server) + .await?; + test.codex.ensure_rollout_materialized().await; + test.codex + .update_thread_metadata( + ThreadMetadataPatch { + daybreak_enabled: Some(only_program != Some(CyberAccessProgram::Standard)), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await?; + let mut created_threads = test.thread_manager.subscribe_thread_created(); + submit(&test, /*program*/ None).await?; + let child = test + .thread_manager + .get_thread(created_threads.recv().await?) + .await?; + wait_for_successful_turn(&child).await; + let before = requests.requests(); + assert_eq!( + before + .iter() + .filter(|request| request.header("x-openai-subagent").is_some()) + .map(|request| request.body_json()["access_programs"].clone()) + .collect::>(), + vec![json!({"cyber": only_program.unwrap_or(CyberAccessProgram::DaybreakBlue)})] + ); + + let accepted_text = "First accepted follow-up."; + child + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: accepted_text.to_owned(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_successful_turn(&child).await; + let requests_before_rejection = requests.requests().len(); + + let expected_error = if let Some(program) = only_program { + child + .update_thread_metadata( + ThreadMetadataPatch { + daybreak_enabled: Some(program == CyberAccessProgram::Standard), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await?; + if program == CyberAccessProgram::Standard { + "does not advertise Daybreak" + } else { + "does not advertise Standard" + } + } else { + let auth = test.thread_manager.auth_manager(); + let previous_owner = auth.auth_owner(); + codex_login::auth::login_with_chatgpt_auth_tokens( + test.home.path(), + "header.e30.changed", + "other-account", + /*chatgpt_plan_type*/ None, + )?; + auth.reload().await; + assert!(!previous_owner.is_current()); + assert_eq!( + auth.auth_cached().and_then(|auth| auth.get_account_id()), + Some("other-account".to_owned()) + ); + + "saved Daybreak choice has changed" + }; + let rejected_text = "Follow-up with a rejected saved choice."; + child + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: rejected_text.to_owned(), + text_elements: Vec::new(), + }])) + .await?; + let completed = + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + assert!( + matches!(&completed, EventMsg::TurnComplete(turn) + if turn.error.as_ref().is_some_and(|error| error.message.contains(expected_error))), + "expected {expected_error}: {completed:?}" + ); + let assert_history = async |thread: &codex_core::CodexThread| { + let history = thread.conversation_history_snapshot().await; + for expected in [accepted_text, rejected_text] { + let input = ContentItem::InputText { + text: expected.to_owned(), + }; + let count = history.items().filter(|item| { + matches!(item, ResponseItem::Message { role, content, .. } if role == "user" && content.contains(&input)) + }).count(); + assert_eq!(count, 1, "{expected}"); + } + }; + assert_history(&child).await; + let (child_id, history) = + super::agents_md::persisted_resume_history(&child, test.thread_store.as_ref()).await?; + child.shutdown_and_wait().await?; + test.thread_manager.remove_thread(&child_id).await; + assert_eq!(requests.requests().len(), requests_before_rejection); + let resumed = test + .thread_manager + .resume_thread_with_history( + test.config.clone(), + history, + test.thread_manager.auth_manager(), + /*parent_trace*/ None, + Default::default(), + ) + .await?; + assert_history(&resumed.thread).await; + resumed.thread.shutdown_and_wait().await?; + test.codex.shutdown_and_wait().await?; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn saved_daybreak_uses_existing_websocket_requests() -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let server = responses::start_websocket_server(vec![ + vec![ + vec![responses::ev_completed("parent-warm")], + vec![ + responses::ev_function_call_with_namespace( + "spawn-worker", + "collaboration", + "spawn_agent", + &json!({"task_name": "worker", "message": "inspect", "fork_turns": "none"}) + .to_string(), + ), + responses::ev_completed("parent-spawn"), + ], + vec![ + responses::ev_assistant_message("parent-done", "Done."), + responses::ev_completed("parent-done"), + ], + ], + vec![ + vec![responses::ev_completed("child-warm")], + vec![ + responses::ev_function_call("pause", "test_tool", "{}"), + responses::ev_completed("child-first"), + ], + vec![responses::ev_completed("child-on")], + vec![responses::ev_completed("child-off")], + ], + ]) + .await; + let release_child = server.gate_response_event(&responses::ev_completed("child-first")); + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model_info_override("gpt-5.4", |model| { + model.available_access_programs = Some(ModelAccessPrograms { + cyber: vec![ + CyberAccessProgram::DaybreakBlue, + CyberAccessProgram::Standard, + ], + }); + model.use_responses_lite = false; + }) + .with_config(|config| { + for feature in [ + Feature::Collab, + Feature::MultiAgentV2, + Feature::DaybreakProgramPropagation, + Feature::StepModelSwitching, + ] { + config + .features + .enable(feature) + .expect("enable saved choice test"); + } + }) + .build_with_websocket_server(&server) + .await?; + test.codex.ensure_rollout_materialized().await; + test.codex + .update_thread_metadata( + ThreadMetadataPatch { + daybreak_enabled: Some(true), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await?; + let mut created_threads = test.thread_manager.subscribe_thread_created(); + submit(&test, /*program*/ None).await?; + let child = test + .thread_manager + .get_thread(created_threads.recv().await?) + .await?; + let EventMsg::TurnStarted(started) = + wait_for_event(&child, |event| matches!(event, EventMsg::TurnStarted(_))).await + else { + unreachable!("waited for turn start"); + }; + server + .wait_for_request(/*connection_index*/ 1, /*request_index*/ 1) + .await; + let (reply, outcome) = oneshot::channel(); + child + .submit(Op::TurnSettings { + turn_id: started.turn_id, + update: TurnSettingsUpdate { + model: Some("gpt-5.5".to_owned()), + ..Default::default() + }, + reply, + }) + .await?; + assert!( + matches!(outcome.await?, TurnSettingsUpdateOutcome::Rejected { reason } if reason.contains("next turn")) + ); + + // A saved toggle during this turn applies to the next turn, not the next request. + child + .update_thread_metadata( + ThreadMetadataPatch { + daybreak_enabled: Some(false), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await?; + release_child.send(()).expect("release child response"); + wait_for_successful_turn(&child).await; + child + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Continue with the new choice.".to_owned(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_successful_turn(&child).await; + + let requests = server.connections(); + assert_eq!(requests.len(), 2); + assert_eq!( + requests[1] + .iter() + .map(|request| { + let body = request.body_json(); + json!([body["access_programs"], body["previous_response_id"]]) + }) + .collect::>(), + vec![ + json!([null, null]), + json!([{"cyber": "daybreak_blue"}, "child-warm"]), + json!([{"cyber": "daybreak_blue"}, "child-first"]), + json!([{"cyber": "standard"}, "child-on"]), + ] + ); + child.shutdown_and_wait().await?; + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} + +#[test_case("multi_agent_v1", true, true, Some(true), CyberAccessProgram::DaybreakBlue, CyberAccessProgram::DaybreakRed; "v1 blue to red")] +#[test_case("multi_agent_v1", true, true, Some(true), CyberAccessProgram::DaybreakRed, CyberAccessProgram::DaybreakBlue; "v1 red to blue")] +#[test_case("collaboration", true, true, Some(true), CyberAccessProgram::DaybreakBlue, CyberAccessProgram::DaybreakRed; "v2 blue to red")] +#[test_case("collaboration", true, true, Some(true), CyberAccessProgram::DaybreakRed, CyberAccessProgram::DaybreakBlue; "v2 red to blue")] +#[test_case("collaboration", true, true, Some(false), CyberAccessProgram::DaybreakBlue, CyberAccessProgram::DaybreakRed; "turn override wins over saved off")] +#[test_case("collaboration", true, true, Some(true), CyberAccessProgram::Standard, CyberAccessProgram::DaybreakRed; "standard override wins over saved on")] +#[test_case("collaboration", true, true, Some(false), CyberAccessProgram::Standard, CyberAccessProgram::DaybreakRed; "standard with saved off")] +#[test_case("collaboration", false, true, Some(true), CyberAccessProgram::DaybreakBlue, CyberAccessProgram::DaybreakRed; "rollout disabled")] +#[test_case("collaboration", true, true, None, CyberAccessProgram::DaybreakBlue, CyberAccessProgram::DaybreakRed; "choice unset")] +#[test_case("collaboration", true, false, Some(true), CyberAccessProgram::DaybreakBlue, CyberAccessProgram::DaybreakRed; "provider opt out")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn saved_daybreak_follows_the_child_model_and_automatic_compaction( + namespace: &'static str, + propagation_enabled: bool, + provider_programs: bool, + saved_choice: Option, + parent_program: CyberAccessProgram, + child_program: CyberAccessProgram, +) -> Result<()> { + core_test_support::skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + let parent_model = "gpt-5.6-sol"; + let child_model = "gpt-5.6-cyber"; + let parent_daybreak = match parent_program { + CyberAccessProgram::Standard => CyberAccessProgram::DaybreakBlue, + program => program, + }; + let is_v2 = namespace == "collaboration"; + let mut spawn = json!({"message": "inspect the repository", "model": child_model}); + if is_v2 { + spawn["task_name"] = json!("worker"); + spawn["fork_turns"] = json!("none"); + } + let summary = json!({"type": "response.output_item.done", "item": { + "type": "compaction", "encrypted_content": "CHILD_SUMMARY" + }}); + let requests = responses::mount_sse_sequence( + &server, + vec![ + responses::sse(vec![ + responses::ev_function_call("before-compact", "test_tool", "{}"), + responses::ev_completed_with_tokens("child-first", /*total_tokens*/ 500_000), + ]), + responses::sse(vec![ + summary.clone(), + responses::ev_completed_with_tokens("compact", /*total_tokens*/ 10), + ]), + responses::sse(vec![responses::ev_completed_with_tokens( + "child-done", + /*total_tokens*/ 80, + )]), + responses::sse(vec![ + summary.clone(), + responses::ev_completed_with_tokens("old-model-compact", /*total_tokens*/ 10), + ]), + responses::sse(vec![ + responses::ev_function_call("before-derived-compact", "test_tool", "{}"), + responses::ev_completed_with_tokens( + "new-model-first", + /*total_tokens*/ 500_000, + ), + ]), + responses::sse(vec![ + summary, + responses::ev_completed_with_tokens( + "derived-program-compact", + /*total_tokens*/ 10, + ), + ]), + responses::sse(vec![responses::ev_completed_with_tokens( + "new-model-done", + /*total_tokens*/ 80, + )]), + responses::sse(vec![responses::ev_completed_with_tokens( + "explicit-done", + /*total_tokens*/ 80, + )]), + ], + ) + .await; + let version = if is_v2 { + MultiAgentVersion::V2 + } else { + MultiAgentVersion::V1 + }; + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model_info_override(parent_model, move |model| { + model.available_access_programs = Some(ModelAccessPrograms { + cyber: vec![parent_daybreak, CyberAccessProgram::Standard], + }); + model.comp_hash = Some("parent-window".to_owned()); + model.multi_agent_version = Some(version); + model.use_responses_lite = false; + }) + .with_model_info_override(child_model, move |model| { + // An enabled choice must work even when this model only accepts Red or Blue. + model.available_access_programs = Some(ModelAccessPrograms { + cyber: if saved_choice == Some(false) + || parent_program == CyberAccessProgram::Standard + { + vec![child_program, CyberAccessProgram::Standard] + } else { + vec![child_program] + }, + }); + model.comp_hash = Some("child-window".to_owned()); + model.multi_agent_version = Some(version); + model.use_responses_lite = false; + }) + .with_model(parent_model) + .with_config(move |config| { + config + .features + .enable(Feature::Collab) + .expect("enable spawn"); + config + .features + .set_enabled(Feature::MultiAgentV2, is_v2) + .expect("set spawn version"); + config + .features + .set_enabled(Feature::DaybreakProgramPropagation, propagation_enabled) + .expect("set propagation"); + config.multi_agent_v2.expose_spawn_agent_model_overrides = true; + config.model_auto_compact_token_limit = Some(200_000); + if !provider_programs { + config.model_provider.capabilities = + Some(codex_model_provider_info::ModelProviderCapabilities { + cyber_access_programs: Some(false), + ..Default::default() + }); + } + }) + .build_with_auto_env(&server) + .await?; + // Parent traffic has its own responses so child compaction is deterministic. + Mock::given(method("POST")) + .and(path("/v1/responses")) + .and(header( + "thread-id", + test.session_configured.thread_id.to_string(), + )) + .respond_with(responses::sse_response(responses::sse(vec![ + responses::ev_function_call_with_namespace( + "spawn-worker", + namespace, + "spawn_agent", + &spawn.to_string(), + ), + responses::ev_completed("parent-spawn"), + ]))) + .up_to_n_times(1) + .with_priority(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/v1/responses")) + .and(header( + "thread-id", + test.session_configured.thread_id.to_string(), + )) + .respond_with(responses::sse_response(responses::sse(vec![ + responses::ev_assistant_message("parent-done", "Done."), + responses::ev_completed("parent-done"), + ]))) + .with_priority(2) + .mount(&server) + .await; + if let Some(enabled) = saved_choice { + test.codex.ensure_rollout_materialized().await; + test.codex + .update_thread_metadata( + ThreadMetadataPatch { + daybreak_enabled: Some(enabled), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await?; + } + let mut created_threads = test.thread_manager.subscribe_thread_created(); + submit(&test, Some(parent_program)).await?; + let child = test + .thread_manager + .get_thread(created_threads.recv().await?) + .await?; + wait_for_successful_turn(&child).await; + + // No explicit program: both sampling and compaction must use the saved choice. + child + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "Continue with the other model.".to_owned(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + model: Some(parent_model.to_owned()), + ..Default::default() + }), + ) + .await?; + let completed = + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + assert!( + matches!(&completed, EventMsg::TurnComplete(turn) if turn.error.is_none()), + "model-switch turn failed: {completed:?}; requests: {:?}", + requests.requests().iter().map(|request| { + let body = request.body_json(); + json!({"model": body["model"], "program": body["access_programs"], "compaction": request.inputs_of_type("compaction_trigger").len()}) + }).collect::>(), + ); + + child + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "Use this explicit choice.".to_owned(), + text_elements: Vec::new(), + }]) + .on_start(TurnStartOptions { + cyber_access_program: Some(CyberAccessProgram::Standard), + ..Default::default() + }), + ) + .await?; + wait_for_successful_turn(&child).await; + child.shutdown_and_wait().await?; + test.codex.shutdown_and_wait().await?; + + let first_turn_program = if parent_program == CyberAccessProgram::Standard { + CyberAccessProgram::Standard + } else { + child_program + }; + let (initial, next) = match (provider_programs, propagation_enabled, saved_choice) { + (false, _, _) => (None, None), + (true, true, Some(true)) => (Some(first_turn_program), Some(parent_daybreak)), + (true, true, Some(false)) => (Some(first_turn_program), Some(CyberAccessProgram::Standard)), + (true, true, None) => (Some(first_turn_program), None), + (true, _, _) => (Some(parent_program), None), + }; + let program_json = |program: Option| { + program + .map(|program| json!({"cyber": program})) + .unwrap_or(json!(null)) + }; + let requests = requests.requests(); + assert_eq!( + requests + .iter() + .map(|request| { + let body = request.body_json(); + json!([body["model"], body["access_programs"]]) + }) + .collect::>(), + vec![ + json!([child_model, program_json(initial)]), + json!([child_model, program_json(initial)]), + json!([child_model, program_json(initial)]), + json!([child_model, program_json(initial)]), + json!([parent_model, program_json(next)]), + json!([parent_model, program_json(next)]), + json!([parent_model, program_json(next)]), + json!([ + parent_model, + program_json(provider_programs.then_some(CyberAccessProgram::Standard)) + ]), + ] + ); + assert_eq!(requests[1].inputs_of_type("compaction_trigger").len(), 1); + assert_eq!(requests[3].inputs_of_type("compaction_trigger").len(), 1); + assert_eq!(requests[5].inputs_of_type("compaction_trigger").len(), 1); + Ok(()) +} + async fn submit(test: &TestCodex, program: Option) -> Result<()> { test.codex .start_or_steer_turn(