//! Handles persistent thread-settings updates and serializes their persistence //! with checkpoints written directly to storage. use super::session::Session; use super::session::SessionSettingsUpdate; use super::step_settings::StepSettingsUpdate; use crate::config::ConstraintResult; use codex_history::RolloutItem; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ThreadSettingsAppliedEvent; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::ThreadSettingsSnapshot; use codex_thread_store::ThreadStoreResult; use std::sync::Arc; use tokio::sync::SemaphorePermit; impl Session { /// Captures and flushes current settings under the shared persistence permit. pub(crate) async fn checkpoint_thread_settings(&self) -> ThreadStoreResult<()> { let _settings_guard = acquire_persistence_lock(self).await; if let Some(live_thread) = self.live_thread() { live_thread .append_items(&[RolloutItem::EventMsg(applied_event(self).await)]) .await?; live_thread.flush().await?; } Ok(()) } } /// Applies standalone thread settings and reports invalid overrides through the /// normal event stream. pub(super) async fn update( session: &Arc, submission_id: String, overrides: ThreadSettingsOverrides, ) { let updates = prepare_update(overrides); if let Err(error) = apply_update(session, submission_id.clone(), updates).await { session .send_event_raw(Event { id: submission_id, msg: EventMsg::Error(ErrorEvent { misalignment: None, message: format!("invalid thread settings override: {error}"), codex_error_info: Some(CodexErrorInfo::BadRequest), }), }) .await; } else { // Standalone settings changes supersede a pending automatic continuation. session.state.lock().await.last_started_turn_id = None; } } /// Converts protocol overrides into the internal settings update shape. pub(super) fn prepare_update(overrides: ThreadSettingsOverrides) -> SessionSettingsUpdate { let ThreadSettingsOverrides { environments, runtime_workspace_roots, profile_workspace_roots, approval_policy, approvals_reviewer, sandbox_policy, permission_profile, active_permission_profile, windows_sandbox_level, model, effort, summary, service_tier, collaboration_mode, personality, disabled_plugin_ids, } = overrides; SessionSettingsUpdate { step_settings: StepSettingsUpdate { model, effort, collaboration_mode, reasoning_summary: summary, service_tier, personality, approval_policy, approvals_reviewer, }, environments, runtime_workspace_roots, profile_workspace_roots, sandbox_policy, permission_profile, active_permission_profile, windows_sandbox_level, disabled_plugin_ids, ..Default::default() } } /// Acquires the shared permit before capturing or changing persistent settings. pub(super) async fn acquire_persistence_lock(session: &Session) -> SemaphorePermit<'_> { session .thread_settings_persistence .acquire() .await .unwrap_or_else(|_| unreachable!("thread settings persistence semaphore is never closed")) } /// Applies persistent settings and emits the resulting thread-owned snapshot. pub(super) async fn apply_update( session: &Session, submission_id: String, updates: SessionSettingsUpdate, ) -> ConstraintResult<()> { let _settings_guard = acquire_persistence_lock(session).await; let commit = session.update_settings(updates).await?; emit_applied(session, submission_id, commit.snapshot).await; Ok(()) } /// Emits the snapshot published by one successful settings update. pub(super) async fn emit_applied( session: &Session, submission_id: String, snapshot: ThreadSettingsSnapshot, ) { let msg = EventMsg::ThreadSettingsApplied(ThreadSettingsAppliedEvent { thread_id: Some(session.thread_id()), thread_settings: snapshot, }); session .send_event_raw_without_materializing_rollout(Event { id: submission_id, msg, }) .await; } /// Builds a current thread-owned snapshot for storage checkpoints. pub(super) async fn applied_event(session: &Session) -> EventMsg { EventMsg::ThreadSettingsApplied(ThreadSettingsAppliedEvent { thread_id: Some(session.thread_id()), thread_settings: session.thread_settings_snapshot().await, }) }