diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index ddf39d9718..79453deeae 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/codex-rs/app-server/src/request_processors/config_processor.rs @@ -72,6 +72,7 @@ const SUPPORTED_EXPERIMENTAL_FEATURE_ENABLEMENT: &[&str] = &[ "auth_elicitation", BACKGROUND_PAGINATED_ROLLOUT_MIGRATION_FEATURE, "codex_apps_mcp_2026_07_28", + "daybreak_program_propagation", "mcp_2026_07_28", "mcp_connection_sharing", "mcp_lazy_stdio_startup", diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index b21b92ca2f..75761a2527 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -50,6 +50,7 @@ use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::mcp::plain_output_text; use codex_protocol::protocol::ThreadHistoryMode; use codex_thread_store::LocalHistoryAccount; +use codex_thread_store::PendingDaybreakPreference; use codex_thread_store::PersistContext; use std::ops::ControlFlow; @@ -3143,17 +3144,14 @@ impl ThreadRequestProcessor { thread_id: ThreadId, include_turns: bool, ) -> Result { - // Read staging first: persistence can consume it while the stored thread is read. - let pending_daybreak_enabled = self - .thread_store - .read_pending_thread_metadata(thread_id) - .await - .map_err(|err| { - ThreadReadViewError::Internal(format!( - "failed to read pending thread metadata: {err}" - )) - })? - .and_then(|metadata| metadata.daybreak_enabled); + let pending_preference = + PendingDaybreakPreference::read(self.thread_store.as_ref(), thread_id) + .await + .map_err(|err| { + ThreadReadViewError::Internal(format!( + "failed to read pending thread metadata: {err}" + )) + })?; let loaded_thread = self.thread_manager.get_thread(thread_id).await.ok(); let mut thread = if include_turns { if let Some(loaded_thread) = loaded_thread.as_ref() { @@ -3207,7 +3205,7 @@ impl ThreadRequestProcessor { ))); }; - thread.daybreak_enabled = thread.daybreak_enabled.or(pending_daybreak_enabled); + thread.daybreak_enabled = pending_preference.resolve(thread.daybreak_enabled); let has_live_in_progress_turn = if let Some(loaded_thread) = loaded_thread.as_ref() { matches!(loaded_thread.agent_status().await, AgentStatus::Running) @@ -3280,14 +3278,7 @@ impl ThreadRequestProcessor { .await { Ok(stored_thread) => Ok(Some(stored_thread)), - Err(ThreadStoreError::InvalidRequest { message }) - if message == format!("no rollout found for thread id {thread_id}") => - { - Ok(None) - } - Err(ThreadStoreError::ThreadNotFound { - thread_id: missing_thread_id, - }) if missing_thread_id == thread_id => Ok(None), + Err(err) if err.is_missing_for_read(thread_id) => Ok(None), Err(ThreadStoreError::InvalidRequest { message }) => { Err(ThreadReadViewError::InvalidRequest(message)) } diff --git a/codex-rs/app-server/tests/suite/v2/experimental_feature_list.rs b/codex-rs/app-server/tests/suite/v2/experimental_feature_list.rs index f89b55c94f..652050a9e2 100644 --- a/codex-rs/app-server/tests/suite/v2/experimental_feature_list.rs +++ b/codex-rs/app-server/tests/suite/v2/experimental_feature_list.rs @@ -282,6 +282,7 @@ async fn experimental_feature_enablement_set_applies_to_global_and_thread_config let enablement = BTreeMap::from([ ("auth_elicitation".to_string(), true), ("mcp_lazy_stdio_startup".to_string(), true), + ("daybreak_program_propagation".to_string(), true), ]); let actual = set_experimental_feature_enablement(&mut mcp, enablement.clone()).await?; assert_eq!(actual.enablement, enablement); diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 6cd281857f..d12af7854c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1061,6 +1061,16 @@ async fn thread_read_loaded_include_turns_reads_store_history_without_rollout_pa items: store_history_items(), }) .await?; + store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + daybreak_enabled: Some(false), + ..Default::default() + }, + include_archived: false, + }) + .await?; let result = client .request(ClientRequest::ThreadRead { @@ -1074,6 +1084,8 @@ async fn thread_read_loaded_include_turns_reads_store_history_without_rollout_pa .expect("thread/read should succeed"); let ThreadReadResponse { thread, .. } = serde_json::from_value(result)?; + assert_eq!(thread.daybreak_enabled, Some(false)); + assert_eq!(thread.status, ThreadStatus::Idle); assert_eq!(turn_user_texts(&thread.turns), vec!["history from store"]); let [ codex_app_server_protocol::ThreadItem { diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 14e34b3f46..c51b05185f 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -971,6 +971,9 @@ "daemon_auto_start": { "type": "boolean" }, + "daybreak_program_propagation": { + "type": "boolean" + }, "default_mode_request_user_input": { "type": "boolean" }, @@ -8035,6 +8038,9 @@ "daemon_auto_start": { "type": "boolean" }, + "daybreak_program_propagation": { + "type": "boolean" + }, "default_mode_request_user_input": { "type": "boolean" }, diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 7449da552d..c026595f61 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -5,6 +5,7 @@ use super::spawn_telemetry::SpawnMeasurements; use super::spawn_telemetry::record_spawn_success; use super::*; use crate::agent::child_config::build_agent_resume_config; +use crate::agent::helper_daybreak; use crate::agent::history::load_agent_model_context; use crate::agent::history::prepare_forked_history; use crate::agent::role::apply_role_to_config; @@ -745,6 +746,54 @@ impl LocalAgentControl { disabled_plugin_ids, ..StartThreadOptions::new(config.clone()) }; + 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 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 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 + && config + .model_provider + .supports_cyber_access_programs(&config.model_provider_id) + && codex_model_provider::provider_uses_first_party_auth_path( + &config.model_provider, + ) + { + return Err(CodexErr::InvalidRequest(error.clone()) + .with_agent_context(AgentErrorContext::ChildStartup)); + } + } else { + start_options.thread_extension_init.insert(intent); + } + } + } let fork_context = match (session_source, options.fork_mode.as_ref()) { (Some(session_source), Some(_)) => { let fork_context_started_at = Instant::now(); diff --git a/codex-rs/core/src/agent/helper_daybreak.rs b/codex-rs/core/src/agent/helper_daybreak.rs new file mode 100644 index 0000000000..430b41d543 --- /dev/null +++ b/codex-rs/core/src/agent/helper_daybreak.rs @@ -0,0 +1,94 @@ +//! Native helpers inherit the spawning turn's saved Daybreak choice. + +pub(crate) mod turn; + +use crate::session::session::Session; +use crate::session::turn_context::TurnContext; +use codex_login::AuthOwner; +use codex_protocol::openai_models::SavedDaybreakChoice; +use codex_thread_store::PendingDaybreakPreference; +use codex_thread_store::ThreadStoreError; + +/// Enrollment is local to a fresh root or a native child admitted at spawn. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Enrollment { + Root, + NativeHelper, +} + +/// The saved choice and the account under which the choice was observed. +#[derive(Clone, Debug)] +pub(crate) struct SavedIntent { + pub(crate) choice: Result, + pub(crate) owner: AuthOwner, +} + +pub(crate) fn eligible_owner(session: &Session) -> Option { + let manager = &session.services.auth_manager; + let credential = manager + .auth_cached() + .filter(codex_login::CodexAuth::is_chatgpt_auth)?; + manager.auth_owner_for(&credential) +} + +pub(crate) async fn spawning_intent( + parent: &Session, + turn: &TurnContext, + enrollment: &Enrollment, + spawn_owner: Option<&AuthOwner>, +) -> Option { + let owner = spawn_owner?; + if turn.config.model_provider_id != codex_model_provider_info::OPENAI_PROVIDER_ID + || !codex_model_provider::provider_uses_first_party_auth_path(turn.provider.info()) + { + return None; + } + if *enrollment == Enrollment::NativeHelper { + // A nested spawn inherits the choice recorded for the sending turn, including + // when a concrete override is forwarded independently in the spawn options. + return turn + .extension_data + .get::() + .map(|intent| intent.as_ref().clone()); + } + 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 { + choice, + owner: owner.clone(), + }) +} + +pub(crate) async fn saved_choice(session: &Session) -> Result, String> { + let thread_id = session.thread_id(); + let read_error = + |error| format!("Could not read the Daybreak preference for thread {thread_id}: {error}"); + let pending = + PendingDaybreakPreference::read(session.services.thread_store.as_ref(), thread_id) + .await + .map_err(read_error)?; + let Some(live) = session.live_thread() else { + return Ok(None); + }; + let stored = match live + .read_thread( + /*include_archived*/ false, /*include_history*/ false, + ) + .await + { + Ok(thread) => thread.daybreak_enabled, + // The live handle already selects the thread for a typed not-found response. + Err(error) + if matches!(error, ThreadStoreError::ThreadNotFound { .. }) + || error.is_missing_for_read(thread_id) => + { + None + } + Err(error) => return Err(read_error(error)), + }; + Ok(pending.resolve(stored)) +} diff --git a/codex-rs/core/src/agent/helper_daybreak/turn.rs b/codex-rs/core/src/agent/helper_daybreak/turn.rs new file mode 100644 index 0000000000..8f899da7b9 --- /dev/null +++ b/codex-rs/core/src/agent/helper_daybreak/turn.rs @@ -0,0 +1,48 @@ +//! Capture the saved Daybreak choice for an admitted native helper turn. +use super::Enrollment; +use super::SavedIntent; +use super::eligible_owner; +use super::saved_choice; +use crate::session::session::Session; +use crate::session::turn_context::TurnContext; +use codex_features::Feature; +use codex_login::AuthOwner; +use codex_protocol::openai_models::SavedDaybreakChoice; + +/// The chat account at admission, captured before preparing the new turn. +pub(crate) struct HelperTurnAdmission(pub(crate) AuthOwner); + +impl HelperTurnAdmission { + pub(crate) fn capture(session: &Session) -> Option { + let enrollment = session.services.thread_extension_data.get::()?; + if *enrollment != Enrollment::NativeHelper { + return None; + } + eligible_owner(session).map(Self) + } + + pub(crate) async fn capture_intent(&self, session: &Session, turn: &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()) + { + return; + } + let (choice, owner) = match saved_choice(session).await { + Ok(Some(enabled)) => (Ok(SavedDaybreakChoice::from(Some(enabled))), self.0.clone()), + Ok(None) => { + let Some(inherited) = session.services.thread_extension_data.get::() + else { + return; + }; + (inherited.choice.clone(), inherited.owner.clone()) + } + Err(_) if turn.cyber_access_program.is_some() => return, + Err(error) => (Err(error), self.0.clone()), + }; + turn.extension_data.insert(SavedIntent { choice, owner }); + } +} diff --git a/codex-rs/core/src/agent/mod.rs b/codex-rs/core/src/agent/mod.rs index dd54f564bb..0b7518b820 100644 --- a/codex-rs/core/src/agent/mod.rs +++ b/codex-rs/core/src/agent/mod.rs @@ -2,6 +2,7 @@ pub(crate) mod agent_resolver; pub(crate) mod api; pub(crate) mod child_config; pub(crate) mod control; +pub(crate) mod helper_daybreak; mod history; pub(crate) mod host; mod registry; diff --git a/codex-rs/core/src/config/runtime_refresh.rs b/codex-rs/core/src/config/runtime_refresh.rs index faf1aa4f8a..d47b722bd9 100644 --- a/codex-rs/core/src/config/runtime_refresh.rs +++ b/codex-rs/core/src/config/runtime_refresh.rs @@ -1,4 +1,4 @@ -//! Resolves refreshable MCP inputs without rebuilding unrelated session configuration. +//! Resolves refreshable MCP inputs and operational controls without rebuilding session configuration. //! Existing sessions refresh ordinary MCP state without gaining enterprise authority. //! Validate the retained backend after applying the refreshable requirements. @@ -92,6 +92,7 @@ impl Config { Feature::McpLazyStdioStartup, Feature::SecretAuthStorage, Feature::McpOAuthRefreshCoordination, + Feature::DaybreakProgramPropagation, ] { if matches!(scope, RuntimeConfigRefresh::UserFiles) { let explicitly_configured = cfg.features.as_ref().is_some_and(|features| { @@ -156,6 +157,15 @@ impl Config { .features .refresh_mcp_features(&features) .map_err(std::io::Error::other)?; + if matches!(scope, RuntimeConfigRefresh::UserFiles) { + config + .features + .set_enabled( + Feature::DaybreakProgramPropagation, + features.enabled(Feature::DaybreakProgramPropagation), + ) + .map_err(std::io::Error::other)?; + } config.mcp_servers = constrain_mcp_servers(servers, layers.requirements().mcp_servers.as_ref()) .map_err(std::io::Error::other)?; diff --git a/codex-rs/core/src/session/config_refresh.rs b/codex-rs/core/src/session/config_refresh.rs index 7464df91e6..ff7a7be638 100644 --- a/codex-rs/core/src/session/config_refresh.rs +++ b/codex-rs/core/src/session/config_refresh.rs @@ -1,4 +1,4 @@ -//! Publishes resolved MCP configuration and host timing rollouts for future turns. +//! Publishes refreshable configuration while preserving unrelated session settings. use crate::McpRefreshReason; use std::sync::Arc; @@ -98,6 +98,10 @@ impl Session { if !matches!(scope, RuntimeConfigRefresh::UserFiles) { for (feature, description) in [ (Feature::Mcp20260728, "MCP protocol"), + ( + Feature::DaybreakProgramPropagation, + "Daybreak helper propagation", + ), (Feature::CodexAppsMcp20260728, "Codex Apps MCP protocol"), ] { if let Err(err) = config diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 04b6314a28..0c73c8e303 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1183,6 +1183,16 @@ impl Session { )), ); } + // Native children receive their enrollment at spawn. A cold resume or durable fork + // has no live attachment, even when the rollout is enabled at startup. + if matches!( + &initial_history, + InitialHistory::New | InitialHistory::Cleared + ) && !session_source.is_non_root_agent() + && config.features.enabled(Feature::DaybreakProgramPropagation) + { + thread_extension_data.insert(crate::agent::helper_daybreak::Enrollment::Root); + } thread_extension_data.insert(crate::context::GuardianReviewEvidence::default()); crate::model_search::SearchOwner::restore(&thread_extension_data, &initial_history); // Kick off independent async setup tasks in parallel to reduce startup latency. diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 63c2ac38aa..2fe58d9dab 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -338,6 +338,7 @@ pub(crate) struct NewTurnContextOptions { pub(crate) final_output_json_schema: Option, pub(crate) prompt_cache_ttl: Option, pub(crate) cyber_access_program: Option, + pub(crate) helper_admission: Option, } /// The context needed for a single turn of the thread. @@ -1489,6 +1490,9 @@ impl Session { turn_context.prompt_cache_ttl = options.prompt_cache_ttl; 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; + } // Tool observers retain this turn's selection even after a later turn starts. if let Some(program) = turn_context.cyber_access_program { turn_context.extension_data.insert(program); diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index b5c9fc5ef3..fb7575936d 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -22,6 +22,7 @@ use super::thread_settings; use super::turn_context::NewTurnContextOptions; use super::turn_context::TurnContext; use crate::WithTurnExtensionData; +use crate::agent::helper_daybreak::turn::HelperTurnAdmission; use crate::context::ContextualUserFragment; use crate::context::GuardianSenderMessages; use crate::state::ActiveTurn; @@ -160,6 +161,12 @@ impl PreparedTurnInputSettings { root_turn_id, cyber_access_program, } = self.start_options; + let helper_admission = + if kind == TurnStartKind::User && turn_trigger.as_deref() != Some("queue") { + HelperTurnAdmission::capture(session) + } else { + None + }; let emit_thread_settings_applied = self.thread_settings_update.is_some(); let _settings_guard = if emit_thread_settings_applied { Some(thread_settings::acquire_persistence_lock(session).await) @@ -174,6 +181,7 @@ impl PreparedTurnInputSettings { final_output_json_schema, prompt_cache_ttl, cyber_access_program, + helper_admission, }; let turn_context = match kind { TurnStartKind::User | TurnStartKind::Recovery => Some( diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 5d15403492..506b17877f 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -27,6 +27,7 @@ use tracing::trace; use tracing::trace_span; use tracing::warn; +use crate::agent::helper_daybreak::turn::HelperTurnAdmission; use crate::codex_thread::BackgroundTerminalInfo; use crate::config::Config; use crate::context::ContextualUserFragment; @@ -535,6 +536,7 @@ impl Session { return; }; + let helper_admission = HelperTurnAdmission::capture(self); self.services .models_manager .refresh_after_auth_change(self.get_config().await.http_client_factory()) @@ -559,9 +561,10 @@ impl Session { }; let (input, mut start_options) = self.input_queue.get_pending_input(&self.active_turn).await; - if !input.iter().any( + let triggered = input.iter().any( |item| matches!(item, TurnInput::InterAgentCommunication(mail) if mail.trigger_turn), - ) { + ); + if !triggered { // Queue-only wakes retain the previous turn context without selecting a new task. start_options = previous_options; start_options.cyber_access_program = self @@ -574,6 +577,7 @@ impl Session { final_output_json_schema: start_options.final_output_json_schema, prompt_cache_ttl: start_options.prompt_cache_ttl, cyber_access_program: start_options.cyber_access_program, + helper_admission: triggered.then_some(helper_admission).flatten(), }; let turn_context = match session_configuration { Some(configuration) => { diff --git a/codex-rs/core/src/thread_manager/helper_daybreak_tests.rs b/codex-rs/core/src/thread_manager/helper_daybreak_tests.rs new file mode 100644 index 0000000000..f754bf63a5 --- /dev/null +++ b/codex-rs/core/src/thread_manager/helper_daybreak_tests.rs @@ -0,0 +1,372 @@ +//! Observe the saved choice at native spawn and admission, before a helper can sample. + +use crate::CodexThread; +use crate::StartThreadOptions; +use crate::ThreadManager; +use crate::agent::helper_daybreak::SavedIntent; +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::config::Config; +use crate::config::test_config; +use crate::session::turn_context::NewTurnContextOptions; +use crate::session::turn_context::TurnContext; +use crate::tools::handlers::multi_agents_common::thread_spawn_source; +use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionFuture; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::TurnLifecycleContributor; +use codex_extension_api::TurnStartInput; +use codex_extension_api::TurnStartPhase; +use codex_features::Feature; +use codex_login::CodexAuth; +use codex_protocol::openai_models::SavedDaybreakChoice; +use codex_protocol::protocol::TurnAbortReason; +use codex_protocol::turn_input::CyberAccessProgram; +use codex_protocol::turn_input::TurnInputRequest; +use codex_protocol::turn_input::TurnStartOptions; +use codex_protocol::user_input::UserInput; +use codex_thread_store::ThreadMetadataPatch; +use core_test_support::PathExt; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use tempfile::tempdir; +use tokio::sync::mpsc; + +struct PauseAdmittedTurn(mpsc::UnboundedSender); + +impl TurnLifecycleContributor for PauseAdmittedTurn { + fn turn_start_phase(&self, _: &ExtensionData) -> TurnStartPhase { + TurnStartPhase::RegularTaskStart + } + + fn on_turn_start<'a>(&'a self, input: TurnStartInput<'a>) -> ExtensionFuture<'a, ()> { + Box::pin(async move { + self.0 + .send(input.turn_id.to_owned()) + .expect("test is waiting for admission"); + std::future::pending().await + }) + } +} + +fn text() -> Vec { + vec![UserInput::Text { + text: "Continue".into(), + text_elements: Vec::new(), + }] +} + +async fn admitted_turn( + turns: &mut mpsc::UnboundedReceiver, + thread: &CodexThread, +) -> Arc { + let id = tokio::time::timeout(std::time::Duration::from_secs(10), turns.recv()) + .await + .expect("regular task should start") + .expect("turn observer"); + thread + .session + .turn_context_for_sub_id(&id) + .await + .expect("turn is paused after admission") +} + +async fn spawn_child( + manager: &ThreadManager, + config: &Config, + parent: &CodexThread, + spawning_turn: &TurnContext, +) -> codex_protocol::error::Result> { + let parent_id = parent.session.thread_id(); + let source = thread_spawn_source( + parent_id, + &parent.session_source, + next_thread_spawn_depth(&parent.session_source), + /*agent_role*/ None, + /*task_name*/ None, + ) + .expect("native child source"); + let child = parent + .session + .services + .local_agent_runtime + .control(parent.session.session_id()) + .spawn_agent_with_metadata( + config.clone(), + text(), + source, + SpawnAgentOptions { + parent_thread_id: Some(parent_id), + parent_turn_id: Some(spawning_turn.sub_id.clone()), + cyber_access_program: spawning_turn.cyber_access_program, + ..Default::default() + }, + ) + .await?; + Ok(manager + .get_thread(child.thread_id) + .await + .expect("loaded child")) +} + +async fn save_choice(thread: &CodexThread, enabled: bool) { + thread.ensure_rollout_materialized().await; + let stored = thread + .update_thread_metadata( + ThreadMetadataPatch { + daybreak_enabled: Some(enabled), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await + .expect("save chat preference"); + assert_eq!(stored.daybreak_enabled, Some(enabled)); +} + +fn saved(store: &ExtensionData) -> Option { + store.get::().map(|intent| { + assert!(intent.owner.is_current()); + intent.choice.clone().expect("read saved choice") + }) +} + +#[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")] +#[test_case::test_case(None, None, Some(CyberAccessProgram::DaybreakRed); "explicit does not invent a choice")] +#[test_case::test_case(Some(true), None, None; "saved on survives toggles and nested spawn")] +#[tokio::test] +async fn native_spawn_captures_the_saved_choice( + stored: Option, + pending: Option, + 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 mut start_options = StartThreadOptions::new(config.clone()); + let thread_store = &manager.state.thread_store; + if let Some(enabled) = pending { + let thread_id = manager.reserve_thread_id(); + thread_store + .stage_pending_thread_metadata( + thread_id, + ThreadMetadataPatch { + daybreak_enabled: Some(enabled), + ..Default::default() + }, + ) + .await + .expect("stage chat preference"); + start_options.reserved_thread_id = Some(thread_id); + } + let root = manager + .start_thread(start_options) + .await + .expect("start root"); + if let Some(enabled) = stored { + save_choice(&root.thread, enabled).await; + } + if pending.is_some() { + // Admitting the first user turn can materialize the parent and consume staging. + let path = root.thread.rollout_path().expect("local rollout path"); + assert!( + !path.exists(), + "parent rollout should still be unmaterialized" + ); + assert_eq!(saved_choice(&root.thread.session).await, Ok(pending)); + } + root.thread + .start_or_steer_turn( + TurnInputRequest::user_input(text()).on_start(TurnStartOptions { + cyber_access_program: explicit, + ..Default::default() + }), + ) + .await + .expect("admit root turn"); + let root_turn = admitted_turn(&mut turns, &root.thread).await; + assert_eq!(saved(&root_turn.extension_data), None); + if stored == Some(true) { + // Keep the writer's live file intact while making metadata reads fail. + let path = root.thread.rollout_path().expect("root rollout"); + 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"); + assert!( + error + .to_string() + .contains("Could not read the Daybreak preference"), + "{error}" + ); + assert_eq!(manager.list_thread_ids().await, vec![root.thread_id]); + 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_turn = admitted_turn(&mut turns, &child).await; + let expected = stored + .or(pending) + .map(|enabled| SavedDaybreakChoice::from(Some(enabled))); + assert_eq!(saved(child.thread_extension_data()), expected); + assert_eq!(saved(&child_turn.extension_data), expected); + assert_eq!(child_turn.cyber_access_program, explicit); + + if stored == Some(true) { + let constructed = child + .session + .new_turn_with_default_settings( + "constructed".into(), + NewTurnContextOptions { + helper_admission: HelperTurnAdmission::capture(&child.session), + ..Default::default() + }, + ) + .await; + assert_eq!( + saved(&constructed.extension_data), + 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_turn = admitted_turn(&mut turns, &sibling).await; + assert_eq!( + saved(&sibling_turn.extension_data), + Some(SavedDaybreakChoice::Off) + ); + sibling + .shutdown_and_wait() + .await + .expect("shut down sibling"); + + let current = root.thread.session.get_config().await; + let mut refreshed = current.as_ref().clone(); + refreshed + .features + .disable(Feature::DaybreakProgramPropagation) + .unwrap(); + assert_eq!( + root.thread + .session + .refresh_runtime_config(current, refreshed) + .await, + crate::ConfigRefreshOutcome::Published + ); + let unenrolled = spawn_child(&manager, &config, &root.thread, &root_turn) + .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!( + unenrolled + .thread_extension_data() + .get::() + .is_none() + ); + unenrolled + .shutdown_and_wait() + .await + .expect("shut down unenrolled child"); + + assert!(manager.remove_thread(&root.thread_id).await.is_some()); + assert!(manager.get_thread(root.thread_id).await.is_err()); + child + .session + .abort_all_tasks(TurnAbortReason::Interrupted) + .await; + + let direct = Some(CyberAccessProgram::DaybreakRed); + child + .start_or_steer_turn( + TurnInputRequest::user_input(text()).on_start(TurnStartOptions { + cyber_access_program: direct, + ..Default::default() + }), + ) + .await + .expect("start explicit helper turn after parent unload"); + let explicit_turn = admitted_turn(&mut turns, &child).await; + assert_eq!(explicit_turn.cyber_access_program, direct); + assert_eq!( + saved(&explicit_turn.extension_data), + Some(SavedDaybreakChoice::On) + ); + + // 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_turn = admitted_turn(&mut turns, &nested).await; + assert_eq!(nested_turn.cyber_access_program, direct); + assert_eq!( + saved(nested.thread_extension_data()), + Some(SavedDaybreakChoice::On) + ); + assert_eq!( + saved(&nested_turn.extension_data), + Some(SavedDaybreakChoice::On) + ); + + child + .session + .abort_all_tasks(TurnAbortReason::Interrupted) + .await; + child + .start_or_steer_turn(TurnInputRequest::user_input(text())) + .await + .expect("admit next helper turn"); + let next = admitted_turn(&mut turns, &child).await; + assert_eq!(next.cyber_access_program, None); + assert_eq!(saved(&next.extension_data), Some(SavedDaybreakChoice::Off)); + assert_eq!( + saved(child.thread_extension_data()), + Some(SavedDaybreakChoice::On) + ); + nested + .shutdown_and_wait() + .await + .expect("shut down nested helper"); + } + child.shutdown_and_wait().await.expect("shut down helper"); + root.thread + .shutdown_and_wait() + .await + .expect("shut down root"); +} diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index c0b813880d..2caafda93d 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1,5 +1,8 @@ #![cfg(test)] +#[path = "thread_manager/helper_daybreak_tests.rs"] +mod helper_daybreak; + use super::*; use crate::ThreadStoreBuilder; use crate::agent::types::SpawnAgentOptions; diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 5c2c64b44d..3b0eeed8c0 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -106,6 +106,8 @@ pub enum Feature { ApiKeyCyberAccessPrograms, /// Enable Daybreak controls and automatic access-program selection in CLI clients. CliDaybreak, + /// Capture the saved Daybreak choice for a native helper and its turns. + DaybreakProgramPropagation, /// Restrict CLI resume of attributed local threads to their creator. CliResumeAccountCheck, /// Removed no-op; use `tui.fullscreen_transcript` instead. @@ -1586,6 +1588,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::Stable, default_enabled: true, }, + FeatureSpec { + id: Feature::DaybreakProgramPropagation, + key: "daybreak_program_propagation", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::EnableRequestCompression, key: "enable_request_compression", diff --git a/codex-rs/thread-store/src/error.rs b/codex-rs/thread-store/src/error.rs index d0ab6d8826..33078efb21 100644 --- a/codex-rs/thread-store/src/error.rs +++ b/codex-rs/thread-store/src/error.rs @@ -81,6 +81,24 @@ pub enum ThreadStoreError { }, } +impl ThreadStoreError { + /// Whether a read found no persisted record for this specific thread. + pub fn is_missing_for_read(&self, requested_thread_id: ThreadId) -> bool { + match self { + Self::ThreadNotFound { thread_id } => *thread_id == requested_thread_id, + // The local store's strict read uses this legacy error for an unmaterialized rollout. + Self::InvalidRequest { message } => { + message == &format!("no rollout found for thread id {requested_thread_id}") + } + Self::Conflict { .. } + | Self::Unsupported { .. } + | Self::Retryable { .. } + | Self::OutcomeUnknown { .. } + | Self::Internal { .. } => false, + } + } +} + #[cfg(test)] #[path = "error_tests.rs"] mod tests; diff --git a/codex-rs/thread-store/src/error_tests.rs b/codex-rs/thread-store/src/error_tests.rs index 4b56fa89bc..680c9da5d3 100644 --- a/codex-rs/thread-store/src/error_tests.rs +++ b/codex-rs/thread-store/src/error_tests.rs @@ -1,6 +1,7 @@ //! Verifies conservative classification at the shared storage error boundary. use super::ThreadStoreError; +use codex_protocol::ThreadId; use pretty_assertions::assert_eq; #[test] @@ -21,3 +22,40 @@ fn unclassified_sqlx_failures_remain_internal() { assert_eq!(message, expected); } } + +#[test] +fn missing_read_requires_the_requested_thread_and_a_known_store_error() { + let requested = ThreadId::new(); + let other = ThreadId::new(); + let missing_rollout = format!("no rollout found for thread id {requested}"); + for error in [ + ThreadStoreError::ThreadNotFound { + thread_id: requested, + }, + ThreadStoreError::InvalidRequest { + message: missing_rollout.clone(), + }, + ] { + assert!(error.is_missing_for_read(requested)); + assert!(!error.is_missing_for_read(other)); + } + for error in [ + ThreadStoreError::InvalidRequest { + message: format!("{missing_rollout} (lookup failed)"), + }, + ThreadStoreError::InvalidRequest { + message: format!("thread {requested} is archived"), + }, + ThreadStoreError::Retryable { + message: missing_rollout.clone(), + }, + ThreadStoreError::Internal { + message: missing_rollout, + }, + ThreadStoreError::Unsupported { + operation: "read_thread", + }, + ] { + assert!(!error.is_missing_for_read(requested), "{error:?}"); + } +} diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index ede0d1022f..495e9b0cd5 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -61,6 +61,7 @@ pub use projects::UpdateProjectParams; pub use projects::UpdatedProject; pub use queue_store::LocalQueueStore; pub use queue_store::QueueStore; +pub use store::PendingDaybreakPreference; pub use store::PersistContext; pub use store::ThreadStore; pub use store::ThreadStoreFuture; diff --git a/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs b/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs index d6973cf3c5..1721596062 100644 --- a/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs +++ b/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs @@ -232,6 +232,63 @@ async fn naming_empty_paginated_thread_propagates_rollout_persistence_failure() ); } +#[tokio::test] +async fn saved_daybreak_preference_survives_materialization_and_prefers_the_stored_choice() { + let (_home, store, _runtime) = store_with_runtime().await; + for (staged, persisted) in [ + (Some(false), Some(true)), + (Some(true), Some(false)), + (None, None), + ] { + let thread_id = ThreadId::new(); + if staged.is_some() { + store + .stage_pending_thread_metadata( + thread_id, + ThreadMetadataPatch { + daybreak_enabled: staged, + ..Default::default() + }, + ) + .await + .expect("stage Daybreak preference"); + } + let pending = crate::PendingDaybreakPreference::read(store.as_ref(), thread_id) + .await + .expect("read staged preference"); + let live = LiveThread::create(store.clone(), create_thread_params(thread_id)) + .await + .expect("create thread"); + live.append_items(&[user_message_item("materialize the saved preference")]) + .await + .expect("materialize thread"); + assert_pending_metadata_consumed(&store, thread_id).await; + assert_eq!(pending.resolve(/*stored*/ None), staged); + + if persisted.is_some() { + store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + daybreak_enabled: persisted, + ..Default::default() + }, + include_archived: false, + }) + .await + .expect("save new preference"); + } + let row = live + .read_thread( + /*include_archived*/ false, /*include_history*/ false, + ) + .await + .expect("read stored preference"); + assert_eq!(row.daybreak_enabled, persisted); + assert_eq!(pending.resolve(row.daybreak_enabled), persisted); + } +} + #[tokio::test] async fn pending_thread_metadata_is_consumed_by_first_metadata_update() { let (_home, store, runtime) = store_with_runtime().await; diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index b5e43dae12..ac0089b166 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -125,6 +125,22 @@ impl PersistContext { } } +/// A saved Daybreak choice staged for a thread before its stored row is materialized. +pub struct PendingDaybreakPreference(Option); + +impl PendingDaybreakPreference { + /// Read staging before the stored row: materialization can consume the staged metadata. + pub async fn read(store: &dyn ThreadStore, thread_id: ThreadId) -> ThreadStoreResult { + let pending = store.read_pending_thread_metadata(thread_id).await?; + Ok(Self(pending.and_then(|patch| patch.daybreak_enabled))) + } + + /// A stored choice, including explicit Off, takes precedence over the staged snapshot. + pub fn resolve(&self, stored: Option) -> Option { + stored.or(self.0) + } +} + /// Storage-neutral thread persistence boundary. pub trait ThreadStore: Any + Send + Sync { /// Return this store as [`Any`] for implementation-owned escape hatches.