use std::borrow::Cow;use std::collections::BTreeMap;use std::collections::HashMap;use std::collections::HashSet;use std::fmt::Debug;use std::path::Path;use std::path::PathBuf;use std::sync::Arc;use std::sync::atomic::AtomicU64;use std::time::SystemTime;use std::time::UNIX_EPOCH;use crate::agent::AgentControl;use crate::agent::AgentStatus;use crate::agent::agent_status_from_event;use crate::agent::status::is_final;use crate::agents_md::LoadedAgentsMd;use crate::attestation::AttestationProvider;use crate::build_available_skills;use crate::compact;use crate::config::ManagedFeatures;use crate::config::resolve_tool_suggest_config_from_layer_stack;use crate::connectors;use crate::context::ApprovedCommandPrefixSaved;use crate::context::AppsInstructions;use crate::context::AvailablePluginsInstructions;use crate::context::AvailableSkillsInstructions;use crate::context::CollaborationModeInstructions;use crate::context::ContextualUserFragment;use crate::context::NetworkRuleSaved;use crate::context::PermissionsInstructions;use crate::context::PersonalitySpecInstructions;use crate::default_skill_metadata_budget;use crate::environment_selection::TurnEnvironmentSnapshot;use crate::exec_policy::ExecPolicyManager;use crate::image_preparation::prepare_response_items;use crate::parse_turn_item;use crate::realtime_conversation::RealtimeConversationManager;use crate::session::turn_context::TurnEnvironment;use crate::session_prefix::format_inter_agent_completion_message;use crate::skills::SkillRenderSideEffects;use crate::skills_load_input_from_config;use crate::turn_metadata::TurnMetadataState;use crate::turn_timing::now_unix_timestamp_ms;use async_channel::Receiver;use async_channel::Sender;use chrono::Local;use chrono::Utc;use codex_analytics::AnalyticsEventsClient;use codex_analytics::SubAgentThreadStartedInput;use codex_analytics::TurnCodexErrorFact;use codex_app_server_protocol::McpServerElicitationRequest;use codex_app_server_protocol::McpServerElicitationRequestParams;use codex_config::types::AuthKeyringBackendKind;use codex_config::types::OAuthCredentialsStoreMode;use codex_exec_server::Environment;use codex_exec_server::EnvironmentManager;use codex_extension_api::ExtensionDataInit;use codex_extension_api::LoadedUserInstructions;use codex_extension_api::PromptSlot;use codex_features::FEATURES;use codex_features::Feature;use codex_features::unstable_features_warning_event;use codex_hooks::Hooks;use codex_hooks::HooksConfig;use codex_login::AuthManager;use codex_login::CodexAuth;use codex_login::auth_env_telemetry::collect_auth_env_telemetry;use codex_login::default_client::originator;use codex_mcp::McpConnectionManager;use codex_mcp::McpResourceClient;use codex_mcp::McpRuntimeContext;use codex_mcp::codex_apps_tools_cache_key;use codex_models_manager::manager::RefreshStrategy;use codex_models_manager::manager::SharedModelsManager;use codex_network_proxy::NetworkProxy;use codex_network_proxy::NetworkProxyAuditMetadata;use codex_network_proxy::normalize_host;use codex_otel::current_span_trace_id;use codex_otel::current_span_w3c_trace_context;use codex_otel::set_parent_from_w3c_trace_context;use codex_protocol::SessionId;use codex_protocol::ThreadId;use codex_protocol::approvals::ElicitationRequestEvent;use codex_protocol::approvals::ExecPolicyAmendment;use codex_protocol::approvals::NetworkPolicyAmendment;use codex_protocol::approvals::NetworkPolicyRuleAction;use codex_protocol::config_types::ApprovalsReviewer;use codex_protocol::config_types::AutoCompactTokenLimitScope;use codex_protocol::config_types::ModeKind;use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE;use codex_protocol::config_types::Settings;use codex_protocol::config_types::WebSearchMode;use codex_protocol::dynamic_tools::DynamicToolResponse;use codex_protocol::dynamic_tools::DynamicToolSpec;use codex_protocol::items::TurnItem;use codex_protocol::items::UserMessageItem;use codex_protocol::mcp::CallToolResult;use codex_protocol::models::ActivePermissionProfile;use codex_protocol::models::AdditionalPermissionProfile;use codex_protocol::models::BaseInstructions;use codex_protocol::models::PermissionProfile;use codex_protocol::models::SandboxEnforcement;use codex_protocol::models::format_allow_prefixes;use codex_protocol::openai_models::ModelInfo;use codex_protocol::openai_models::ModelPreset;use codex_protocol::permissions::FileSystemSandboxPolicy;use codex_protocol::permissions::NetworkSandboxPolicy;use codex_protocol::protocol::AdditionalContextEntry;use codex_protocol::protocol::FileChange;use codex_protocol::protocol::HasLegacyEvent;use codex_protocol::protocol::InterAgentCommunication;use codex_protocol::protocol::ItemCompletedEvent;use codex_protocol::protocol::ItemStartedEvent;use codex_protocol::protocol::MultiAgentVersion;use codex_protocol::protocol::RawResponseItemEvent;use codex_protocol::protocol::ReviewRequest;use codex_protocol::protocol::RolloutItem;use codex_protocol::protocol::SessionSource;use codex_protocol::protocol::SubAgentSource;use codex_protocol::protocol::ThreadSource;use codex_protocol::protocol::TurnAbortReason;use codex_protocol::protocol::TurnContextItem;use codex_protocol::protocol::TurnContextNetworkItem;use codex_protocol::protocol::TurnEnvironmentSelection;use codex_protocol::protocol::TurnEnvironmentSelections;use codex_protocol::protocol::W3cTraceContext;use codex_protocol::request_permissions::PermissionGrantScope;use codex_protocol::request_permissions::RequestPermissionProfile;use codex_protocol::request_permissions::RequestPermissionsArgs;use codex_protocol::request_permissions::RequestPermissionsEvent;use codex_protocol::request_permissions::RequestPermissionsResponse;use codex_protocol::request_user_input::RequestUserInputArgs;use codex_protocol::request_user_input::RequestUserInputResponse;use codex_rmcp_client::ElicitationResponse;use codex_rollout::state_db;use codex_rollout_trace::AgentResultTracePayload;use codex_rollout_trace::ThreadStartedTraceMetadata;use codex_rollout_trace::ThreadTraceContext;use codex_sandboxing::policy_transforms::intersect_permission_profiles;use codex_shell_command::parse_command::parse_command;use codex_terminal_detection::user_agent;use codex_thread_store::CreateThreadParams;use codex_thread_store::LiveThread;use codex_thread_store::LiveThreadInitGuard;use codex_thread_store::LocalThreadStore;use codex_thread_store::ReadThreadParams;use codex_thread_store::ResumeThreadParams;use codex_thread_store::ThreadPersistenceMetadata;use codex_thread_store::ThreadStore;use codex_utils_output_truncation::TruncationPolicy;use codex_utils_path_uri::PathUri;use futures::future::BoxFuture;use futures::future::Shared;use futures::prelude::*;use rmcp::model::ElicitationCapability;use rmcp::model::FormElicitationCapability;use rmcp::model::ListResourceTemplatesResult;use rmcp::model::ListResourcesResult;use rmcp::model::PaginatedRequestParams;use rmcp::model::ReadResourceRequestParams;use rmcp::model::ReadResourceResult;use rmcp::model::RequestId;use rmcp::model::UrlElicitationCapability;use serde_json::Value;use tokio::sync::Mutex;use tokio::sync::RwLock;use tokio::sync::oneshot;use tokio::sync::watch;use tokio::task::JoinHandle;use tokio_util::sync::CancellationToken;use toml::Value as TomlValue;use tracing::Instrument;use tracing::debug;use tracing::error;use tracing::info;use tracing::info_span;use tracing::instrument;use tracing::warn;use uuid::Uuid;use crate::client::ModelClient;use crate::codex_thread::ThreadConfigSnapshot;#[cfg(test)]use crate::compact::collect_user_messages;use crate::config::Config;use crate::config::Constrained;use crate::config::ConstraintResult;use crate::config::PermissionProfileSnapshot;use crate::config::PermissionProfileState;use crate::config::StartedNetworkProxy;use crate::config::resolve_web_search_mode_for_turn;use crate::context_manager::ContextManager;use crate::thread_rollout_truncation::initial_history_has_prior_user_turns;use codex_config::CONFIG_TOML_FILE;use codex_config::ConfigLayerSource;use codex_config::ConfigLayerStackOrdering;use codex_config::types::McpServerConfig;use codex_model_provider_info::ModelProviderInfo;use codex_protocol::config_types::ShellEnvironmentPolicy;use codex_protocol::error::CodexErr;use codex_protocol::error::Result as CodexResult;#[cfg(test)]use codex_protocol::exec_output::StreamOutput;mod config_lock;mod handlers;mod inject;mod input_queue;mod mcp;mod multi_agents;mod review;mod rollout_reconstruction;#[allow(clippy::module_inception)]pub(crate) mod session;mod token_budget;pub(crate) mod turn;pub(crate) mod turn_context;use self::config_lock::export_config_lock_if_configured;use self::config_lock::validate_config_lock_if_configured;#[cfg(test)]use self::handlers::submission_dispatch_span;use self::handlers::submission_loop;pub(crate) use self::input_queue::InputQueueActivity;pub(crate) use self::input_queue::TurnInput;pub(crate) use self::input_queue::TurnInputQueue;use self::review::spawn_review_thread;use self::session::AppServerClientMetadata;use self::session::Session;use self::session::SessionConfiguration;pub(crate) use self::session::SessionSettingsUpdate;#[cfg(test)]use self::turn::AssistantMessageStreamParsers;#[cfg(test)]use self::turn::collect_explicit_app_ids_from_skill_items;use self::turn::realtime_text_for_event;use self::turn_context::TurnContext;use self::turn_context::TurnSkillsContext;#[cfg(test)]mod rollout_reconstruction_tests;#[derive(Debug, PartialEq)]pub enum SteerInputError {    NoActiveTurn(Vec<UserInput>),    ExpectedTurnMismatch { expected: String, actual: String },    ActiveTurnNotSteerable { turn_kind: NonSteerableTurnKind },    EmptyInput,}impl SteerInputError {    fn to_error_event(&self) -> ErrorEvent {        match self {            Self::NoActiveTurn(_) => ErrorEvent {                message: "no active turn to steer".to_string(),                codex_error_info: Some(CodexErrorInfo::BadRequest),            },            Self::ExpectedTurnMismatch { expected, actual } => ErrorEvent {                message: format!("expected active turn id `{expected}` but found `{actual}`"),                codex_error_info: Some(CodexErrorInfo::BadRequest),            },            Self::ActiveTurnNotSteerable { turn_kind } => {                let turn_kind_label = match turn_kind {                    NonSteerableTurnKind::Review => "review",                    NonSteerableTurnKind::Compact => "compact",                };                ErrorEvent {                    message: format!("cannot steer a {turn_kind_label} turn"),                    codex_error_info: Some(CodexErrorInfo::ActiveTurnNotSteerable {                        turn_kind: *turn_kind,                    }),                }            }            Self::EmptyInput => ErrorEvent {                message: "input must not be empty".to_string(),                codex_error_info: Some(CodexErrorInfo::BadRequest),            },        }    }}/// Notes from the previous real user turn.////// Conceptually this is the same role that `previous_model` used to fill, but/// it can carry other prior-turn settings that matter when constructing/// sensible state-change diffs or full-context reinjection, such as model/// switches, compaction compatibility, or detecting a prior/// `realtime_active -> false` transition.#[derive(Clone, Debug, PartialEq, Eq)]pub(crate) struct PreviousTurnSettings {    pub(crate) model: String,    pub(crate) comp_hash: Option<String>,    pub(crate) realtime_active: Option<bool>,}#[cfg(test)]use crate::SkillLoadOutcome;#[cfg(test)]use crate::SkillMetadata;use crate::SkillsManager;use crate::agents_md::load_project_instructions;use crate::exec_policy::ExecPolicyUpdateError;use crate::guardian::GuardianReviewSessionManager;use crate::mcp::McpManager;use crate::network_policy_decision::execpolicy_network_rule_amendment;use crate::rollout::map_session_init_error;use crate::session_startup_prewarm::SessionStartupPrewarmHandle;use crate::shell;use crate::state::AutoCompactWindowSnapshot;use crate::state::PendingRequestPermissions;use crate::state::SessionServices;use crate::state::SessionState;#[cfg(test)]use crate::stream_events_utils::HandleOutputCtx;#[cfg(test)]use crate::stream_events_utils::handle_output_item_done;use crate::tasks::ReviewTask;use crate::tools::network_approval::NetworkApprovalService;use crate::tools::network_approval::build_blocked_request_observer;use crate::tools::network_approval::build_network_policy_decider;#[cfg(test)]use crate::tools::parallel::ToolCallRuntime;use crate::tools::sandboxing::ApprovalStore;use crate::turn_timing::TurnTimingState;use crate::turn_timing::record_turn_ttfm_metric;use crate::unified_exec::UnifiedExecProcessManager;use crate::windows_sandbox::WindowsSandboxLevelExt;use codex_core_plugins::PluginsManager;use codex_git_utils::get_git_repo_root;use codex_mcp::McpConfig;use codex_mcp::compute_auth_statuses;use codex_mcp::effective_mcp_servers_from_configured;use codex_mcp::host_owned_codex_apps_enabled;use codex_otel::SessionTelemetry;use codex_otel::THREAD_STARTED_METRIC;use codex_otel::TelemetryAuthMode;use codex_protocol::config_types::CollaborationMode;use codex_protocol::config_types::Personality;use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;use codex_protocol::config_types::WindowsSandboxLevel;use codex_protocol::models::LocalImagePreparation;use codex_protocol::models::ResponseInputItem;use codex_protocol::models::ResponseItem;use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;use codex_protocol::protocol::AskForApproval;use codex_protocol::protocol::CodexErrorInfo;use codex_protocol::protocol::CompactedItem;use codex_protocol::protocol::DeprecationNoticeEvent;use codex_protocol::protocol::ErrorEvent;use codex_protocol::protocol::Event;use codex_protocol::protocol::EventMsg;use codex_protocol::protocol::ExecApprovalRequestEvent;use codex_protocol::protocol::InitialHistory;use codex_protocol::protocol::McpServerRefreshConfig;use codex_protocol::protocol::ModelRerouteEvent;use codex_protocol::protocol::ModelRerouteReason;use codex_protocol::protocol::ModelVerification;use codex_protocol::protocol::ModelVerificationEvent;use codex_protocol::protocol::NetworkApprovalContext;use codex_protocol::protocol::NonSteerableTurnKind;use codex_protocol::protocol::Op;use codex_protocol::protocol::RateLimitSnapshot;use codex_protocol::protocol::RequestUserInputEvent;use codex_protocol::protocol::ReviewDecision;use codex_protocol::protocol::SandboxPolicy;use codex_protocol::protocol::SessionConfiguredEvent;use codex_protocol::protocol::SessionNetworkProxyRuntime;use codex_protocol::protocol::StreamErrorEvent;use codex_protocol::protocol::Submission;use codex_protocol::protocol::ThreadMemoryMode;use codex_protocol::protocol::TokenCountEvent;use codex_protocol::protocol::TokenUsage;use codex_protocol::protocol::TokenUsageInfo;use codex_protocol::protocol::TurnModerationMetadataEvent;use codex_protocol::protocol::WarningEvent;use codex_protocol::user_input::UserInput;use codex_tools::ToolEnvironmentMode;use codex_tools::UnifiedExecShellMode;use codex_utils_absolute_path::AbsolutePathBuf;#[cfg(test)]use codex_utils_stream_parser::ProposedPlanSegment;/// The high-level interface to the Codex system./// It operates as a queue pair where you send submissions and receive events.pub struct Codex {    pub(crate) tx_sub: Sender<Submission>,    pub(crate) rx_event: Receiver<Event>,    // Last known status of the agent.    pub(crate) agent_status: watch::Receiver<AgentStatus>,    pub(crate) session: Arc<Session>,    // Shared future for the background submission loop completion so multiple    // callers can wait for shutdown.    pub(crate) session_loop_termination: SessionLoopTermination,}pub(crate) type SessionLoopTermination = Shared<BoxFuture<'static, ()>>;/// Wrapper returned by [`Codex::spawn`] containing the spawned [`Codex`] and/// the unique session id.pub struct CodexSpawnOk {    pub codex: Codex,    pub thread_id: ThreadId,}pub(crate) struct CodexSpawnArgs {    pub(crate) config: Config,    pub(crate) user_instructions: LoadedUserInstructions,    pub(crate) installation_id: String,    pub(crate) auth_manager: Arc<AuthManager>,    pub(crate) models_manager: SharedModelsManager,    pub(crate) environment_manager: Arc<EnvironmentManager>,    pub(crate) skills_manager: Arc<SkillsManager>,    pub(crate) plugins_manager: Arc<PluginsManager>,    pub(crate) mcp_manager: Arc<McpManager>,    pub(crate) extensions: Arc<codex_extension_api::ExtensionRegistry<crate::config::Config>>,    pub(crate) conversation_history: InitialHistory,    pub(crate) session_source: SessionSource,    pub(crate) forked_from_thread_id: Option<ThreadId>,    pub(crate) parent_thread_id: Option<ThreadId>,    pub(crate) thread_source: Option<ThreadSource>,    pub(crate) agent_control: AgentControl,    pub(crate) dynamic_tools: Vec<DynamicToolSpec>,    pub(crate) metrics_service_name: Option<String>,    pub(crate) inherited_exec_policy: Option<Arc<ExecPolicyManager>>,    pub(crate) inherited_environments: Option<TurnEnvironmentSnapshot>,    /// Parent rollout trace used only to derive fresh spawned child traces.    ///    /// Root sessions and non-thread-spawn subagents pass a disabled context;    /// `Session::new` creates the root trace itself when rollout tracing is enabled.    pub(crate) parent_rollout_thread_trace: ThreadTraceContext,    pub(crate) user_shell_override: Option<shell::Shell>,    pub(crate) parent_trace: Option<W3cTraceContext>,    pub(crate) environment_selections: Vec<TurnEnvironmentSelection>,    pub(crate) thread_extension_init: ExtensionDataInit,    pub(crate) analytics_events_client: Option<AnalyticsEventsClient>,    pub(crate) thread_store: Arc<dyn ThreadStore>,    pub(crate) attestation_provider: Option<Arc<dyn AttestationProvider>>,    pub(crate) inherited_multi_agent_version: Option<MultiAgentVersion>,}pub(crate) fn resolve_multi_agent_version(    conversation_history: &InitialHistory,    inherited_multi_agent_version: Option<MultiAgentVersion>,) -> Option<MultiAgentVersion> {    if inherited_multi_agent_version == Some(MultiAgentVersion::Disabled) {        return Some(MultiAgentVersion::Disabled);    }    conversation_history        .get_multi_agent_version()        .or(inherited_multi_agent_version)        .or(match conversation_history {            InitialHistory::New | InitialHistory::Cleared => None,            // Threads created before runtime metadata existed keep the legacy V1 tool surface.            InitialHistory::Resumed(_) | InitialHistory::Forked(_) => Some(MultiAgentVersion::V1),        })}pub(crate) const INITIAL_SUBMIT_ID: &str = "";pub(crate) const SUBMISSION_CHANNEL_CAPACITY: usize = 512;const CYBER_VERIFY_URL: &str = "https://chatgpt.com/cyber";const CYBER_SAFETY_URL: &str = "https://developers.openai.com/codex/concepts/cyber-safety";impl Codex {    /// Spawn a new [`Codex`] and initialize the session.    pub(crate) async fn spawn(args: CodexSpawnArgs) -> CodexResult<CodexSpawnOk> {        let parent_trace = match args.parent_trace {            Some(trace) => {                if codex_otel::context_from_w3c_trace_context(&trace).is_some() {                    Some(trace)                } else {                    warn!("ignoring invalid thread spawn trace carrier");                    None                }            }            None => None,        };        let thread_spawn_span = info_span!("thread_spawn", otel.name = "thread_spawn");        if let Some(trace) = parent_trace.as_ref() {            let _ = set_parent_from_w3c_trace_context(&thread_spawn_span, trace);        }        Self::spawn_internal(CodexSpawnArgs {            parent_trace,            ..args        })        .instrument(thread_spawn_span)        .await    }    async fn spawn_internal(args: CodexSpawnArgs) -> CodexResult<CodexSpawnOk> {        let CodexSpawnArgs {            mut config,            user_instructions,            installation_id,            auth_manager,            models_manager,            environment_manager,            skills_manager,            plugins_manager,            mcp_manager,            extensions,            conversation_history,            session_source,            forked_from_thread_id,            parent_thread_id,            thread_source,            agent_control,            dynamic_tools,            metrics_service_name,            user_shell_override,            inherited_exec_policy,            inherited_environments,            parent_rollout_thread_trace,            parent_trace: _,            environment_selections,            thread_extension_init,            analytics_events_client,            thread_store,            attestation_provider,            inherited_multi_agent_version,        } = args;        let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);        let (tx_event, rx_event) = async_channel::unbounded();        let LoadedUserInstructions {            instructions: user_instructions,            warnings: user_instruction_provider_warnings,        } = user_instructions;        // TODO(anp) pull startup_warnings out of Config        config            .startup_warnings            .extend(user_instruction_provider_warnings);        let exec_policy = if crate::guardian::is_guardian_reviewer_source(&session_source) {            // Guardian review should rely on the built-in shell safety checks,            // not on caller-provided exec-policy rules that could shape the            // reviewer or silently auto-approve commands.            Arc::new(ExecPolicyManager::default())        } else if let Some(exec_policy) = &inherited_exec_policy {            Arc::clone(exec_policy)        } else {            Arc::new(                ExecPolicyManager::load(&config.config_layer_stack)                    .await                    .map_err(|err| CodexErr::Fatal(format!("failed to load rules: {err}")))?,            )        };        let config = Arc::new(config);        let refresh_strategy = if session_source.is_non_root_agent() {            codex_models_manager::manager::RefreshStrategy::Offline        } else {            codex_models_manager::manager::RefreshStrategy::OnlineIfUncached        };        if config.model.is_none()            || !matches!(                refresh_strategy,                codex_models_manager::manager::RefreshStrategy::Offline            )        {            let _ = models_manager.list_models(refresh_strategy).await;        }        let model = models_manager            .get_default_model(&config.model, refresh_strategy)            .await;        // Resolve base instructions for the session. Priority order:        // 1. config.base_instructions override        // 2. conversation history => session_meta.base_instructions        // 3. base_instructions for current model        let model_info = models_manager            .get_model_info(model.as_str(), &config.to_models_manager_config())            .await;        let multi_agent_version =            resolve_multi_agent_version(&conversation_history, inherited_multi_agent_version);        config            .validate_multi_agent_v2_config()            .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?;        let base_instructions = config            .base_instructions            .clone()            .or_else(|| conversation_history.get_base_instructions().map(|s| s.text))            .unwrap_or_else(|| model_info.get_model_instructions(config.personality));        // Dynamic tools are defined at thread start and persisted in rollout session metadata.        let dynamic_tools = if dynamic_tools.is_empty() {            conversation_history.get_dynamic_tools().unwrap_or_default()        } else {            dynamic_tools        };        // TODO (aibrahim): Consolidate config.model and config.model_reasoning_effort into config.collaboration_mode        // to avoid extracting these fields separately and constructing CollaborationMode here.        let collaboration_mode = CollaborationMode {            mode: ModeKind::Default,            settings: Settings {                model: model.clone(),                reasoning_effort: config.model_reasoning_effort.clone(),                developer_instructions: None,            },        };        let service_tier = get_service_tier(            config.service_tier.clone(),            config.features.enabled(Feature::FastMode),            &model_info,        );        let session_configuration = SessionConfiguration {            provider: config.model_provider.clone(),            collaboration_mode,            model_reasoning_summary: config.model_reasoning_summary,            service_tier,            developer_instructions: config.developer_instructions.clone(),            loaded_agents_md: None,            personality: config.personality,            base_instructions,            compact_prompt: config.compact_prompt.clone(),            approval_policy: config.permissions.approval_policy.clone(),            approvals_reviewer: config.approvals_reviewer,            permission_profile_state: session_permission_profile_state_from_config(&config)?,            windows_sandbox_level: WindowsSandboxLevel::from_config(&config),            environments: TurnEnvironmentSelections::new(                config.cwd.clone(),                environment_selections,            ),            workspace_roots: config.workspace_roots.clone(),            codex_home: config.codex_home.clone(),            thread_name: None,            original_config_do_not_use: Arc::clone(&config),            metrics_service_name,            app_server_client_name: None,            app_server_client_version: None,            session_source,            forked_from_thread_id,            parent_thread_id,            thread_source,            dynamic_tools,            user_shell_override,        };        // Generate a unique ID for the lifetime of this Codex session.        let session_source_clone = session_configuration.session_source.clone();        let (agent_status_tx, agent_status_rx) = watch::channel(AgentStatus::PendingInit);        let session = Box::pin(Session::new(            session_configuration,            config.clone(),            user_instructions,            installation_id,            auth_manager.clone(),            models_manager.clone(),            exec_policy,            tx_event.clone(),            agent_status_tx.clone(),            conversation_history,            session_source_clone,            skills_manager,            plugins_manager,            mcp_manager.clone(),            extensions,            thread_extension_init,            agent_control,            environment_manager,            inherited_environments,            analytics_events_client,            thread_store,            parent_rollout_thread_trace,            attestation_provider,            multi_agent_version,        ))        .await        .map_err(|e| {            error!("Failed to create session: {e:#}");            map_session_init_error(&e, &config.codex_home)        })?;        let thread_id = session.thread_id;        // This task will run until Op::Shutdown is received.        let session_for_loop = Arc::clone(&session);        let session_loop_handle = tokio::spawn(async move {            submission_loop(session_for_loop, config, rx_sub)                .instrument(info_span!("session_loop", thread_id = %thread_id))                .await;        });        let codex = Codex {            tx_sub,            rx_event,            agent_status: agent_status_rx,            session,            session_loop_termination: session_loop_termination_from_handle(session_loop_handle),        };        Ok(CodexSpawnOk { codex, thread_id })    }    /// Submit the `op` wrapped in a `Submission` with a unique ID.    pub async fn submit(&self, op: Op) -> CodexResult<String> {        self.submit_with_trace(op, /*trace*/ None).await    }    pub async fn submit_with_trace(        &self,        op: Op,        trace: Option<W3cTraceContext>,    ) -> CodexResult<String> {        let id = Uuid::now_v7().to_string();        let sub = Submission {            id: id.clone(),            op,            client_user_message_id: None,            trace,        };        self.submit_with_id(sub).await?;        Ok(id)    }    pub async fn submit_user_input_with_client_user_message_id(        &self,        op: Op,        trace: Option<W3cTraceContext>,        client_user_message_id: Option<String>,    ) -> CodexResult<String> {        debug_assert!(matches!(op, Op::UserInput { .. }));        let id = Uuid::now_v7().to_string();        let sub = Submission {            id: id.clone(),            op,            client_user_message_id,            trace,        };        self.submit_with_id(sub).await?;        Ok(id)    }    /// Use sparingly: prefer `submit()` so Codex is responsible for generating    /// unique IDs for each submission.    pub async fn submit_with_id(&self, mut sub: Submission) -> CodexResult<()> {        if sub.trace.is_none() {            sub.trace = current_span_w3c_trace_context();        }        self.tx_sub            .send(sub)            .await            .map_err(|_| CodexErr::InternalAgentDied)?;        Ok(())    }    /// Persist a thread-level memory mode update for the active session.    ///    /// This is a local-only operation that updates rollout metadata directly    /// and does not involve the model.    pub async fn set_thread_memory_mode(        &self,        mode: codex_protocol::protocol::ThreadMemoryMode,    ) -> anyhow::Result<()> {        handlers::persist_thread_memory_mode_update(&self.session, mode).await    }    pub async fn shutdown_and_wait(&self) -> CodexResult<()> {        let session_loop_termination = self.session_loop_termination.clone();        match self.submit(Op::Shutdown).await {            Ok(_) => {}            Err(CodexErr::InternalAgentDied) => {}            Err(err) => return Err(err),        }        session_loop_termination.await;        Ok(())    }    pub async fn next_event(&self) -> CodexResult<Event> {        let event = self            .rx_event            .recv()            .await            .map_err(|_| CodexErr::InternalAgentDied)?;        Ok(event)    }    pub async fn steer_input(        &self,        input: Vec<UserInput>,        additional_context: BTreeMap<String, AdditionalContextEntry>,        expected_turn_id: Option<&str>,        client_user_message_id: Option<String>,        responsesapi_client_metadata: Option<HashMap<String, String>>,    ) -> Result<String, SteerInputError> {        self.session            .steer_input(                input,                additional_context,                expected_turn_id,                client_user_message_id,                responsesapi_client_metadata,            )            .await    }    pub(crate) async fn set_app_server_client_info(        &self,        app_server_client_name: Option<String>,        app_server_client_version: Option<String>,        mcp_elicitations_auto_deny: bool,    ) -> ConstraintResult<()> {        self.session            .update_settings(SessionSettingsUpdate {                app_server_client_name,                app_server_client_version,                ..Default::default()            })            .await?;        let mcp_connection_manager = self.session.services.mcp_connection_manager.load_full();        mcp_connection_manager.set_elicitations_auto_deny(mcp_elicitations_auto_deny);        Ok(())    }    pub(crate) async fn agent_status(&self) -> AgentStatus {        self.agent_status.borrow().clone()    }    pub(crate) async fn thread_config_snapshot(&self) -> ThreadConfigSnapshot {        let state = self.session.state.lock().await;        state.session_configuration.thread_config_snapshot()    }    pub(crate) async fn instruction_sources(&self) -> Vec<AbsolutePathBuf> {        let state = self.session.state.lock().await;        state            .session_configuration            .loaded_agents_md            .as_ref()            .map_or_else(Vec::new, |instructions| {                instructions.sources().cloned().collect()            })    }    pub(crate) async fn thread_environment_selections(&self) -> Vec<TurnEnvironmentSelection> {        let state = self.session.state.lock().await;        state            .session_configuration            .environment_selections()            .to_vec()    }    pub(crate) fn state_db(&self) -> Option<state_db::StateDbHandle> {        self.session.state_db()    }    pub(crate) fn enabled(&self, feature: Feature) -> bool {        self.session.enabled(feature)    }}fn get_service_tier(    configured_service_tier: Option<String>,    fast_mode_enabled: bool,    model_info: &ModelInfo,) -> Option<String> {    if !fast_mode_enabled {        return None;    }    configured_service_tier.filter(|service_tier| {        service_tier == SERVICE_TIER_DEFAULT_REQUEST_VALUE            || model_info.supports_service_tier(service_tier)    })}fn session_permission_profile_state_from_config(    config: &Config,) -> CodexResult<PermissionProfileState> {    Ok(config.permissions.permission_profile_state().clone())}#[cfg(test)]pub(crate) fn completed_session_loop_termination() -> SessionLoopTermination {    futures::future::ready(()).boxed().shared()}pub(crate) fn session_loop_termination_from_handle(    handle: JoinHandle<()>,) -> SessionLoopTermination {    async move {        let _ = handle.await;    }    .boxed()    .shared()}async fn thread_title_from_thread_store(    live_thread: Option<&LiveThread>,    thread_store: &Arc<dyn ThreadStore>,    conversation_id: ThreadId,) -> Option<String> {    let thread = match live_thread {        Some(live_thread) => {            live_thread                .read_thread(                    /*include_archived*/ true, /*include_history*/ false,                )                .await        }        None => {            thread_store                .read_thread(ReadThreadParams {                    thread_id: conversation_id,                    include_archived: true,                    include_history: false,                })                .await        }    }    .ok()?;    let title = thread.name.as_deref()?.trim();    (!title.is_empty() && thread.preview.trim() != title).then(|| title.to_string())}impl Session {    pub(crate) async fn app_server_client_metadata(&self) -> AppServerClientMetadata {        let state = self.state.lock().await;        AppServerClientMetadata {            client_name: state.session_configuration.app_server_client_name.clone(),            client_version: state                .session_configuration                .app_server_client_version                .clone(),        }    }    fn managed_network_proxy_active_for_permission_profile(        permission_profile: &PermissionProfile,    ) -> bool {        !matches!(permission_profile, PermissionProfile::Disabled)    }    /// Builds the `x-codex-beta-features` header value for this session.    ///    /// `ModelClient` is session-scoped and intentionally does not depend on the full `Config`, so    /// we precompute the comma-separated list of enabled experimental feature keys at session    /// creation time and thread it into the client.    fn build_model_client_beta_features_header(config: &Config) -> Option<String> {        let beta_features_header = FEATURES            .iter()            .filter_map(|spec| {                let advertise_in_model_client_header =                    spec.stage.experimental_menu_description().is_some()                        || spec.id == Feature::RemoteCompactionV2;                if advertise_in_model_client_header && config.features.enabled(spec.id) {                    Some(spec.key)                } else {                    None                }            })            .collect::<Vec<_>>()            .join(",");        if beta_features_header.is_empty() {            None        } else {            Some(beta_features_header)        }    }    async fn start_managed_network_proxy(        spec: &crate::config::NetworkProxySpec,        exec_policy: &codex_execpolicy::Policy,        permission_profile: &PermissionProfile,        network_policy_decider: Option<Arc<dyn codex_network_proxy::NetworkPolicyDecider>>,        blocked_request_observer: Option<Arc<dyn codex_network_proxy::BlockedRequestObserver>>,        managed_network_requirements_enabled: bool,        audit_metadata: NetworkProxyAuditMetadata,    ) -> anyhow::Result<(StartedNetworkProxy, SessionNetworkProxyRuntime)> {        let spec = spec            .with_exec_policy_network_rules(exec_policy)            .map_err(|err| {                tracing::warn!(                    "failed to apply execpolicy network rules to managed proxy; continuing with configured network policy: {err}"                );                err            })            .unwrap_or_else(|_| spec.clone());        let network_proxy = spec            .start_proxy(                permission_profile,                network_policy_decider,                blocked_request_observer,                managed_network_requirements_enabled,                audit_metadata,            )            .await            .map_err(|err| anyhow::anyhow!("failed to start managed network proxy: {err}"))?;        let session_network_proxy = {            let proxy = network_proxy.proxy();            SessionNetworkProxyRuntime {                http_addr: proxy.http_addr().to_string(),                socks_addr: proxy.socks_addr().to_string(),            }        };        Ok((network_proxy, session_network_proxy))    }    async fn refresh_managed_network_proxy_for_current_permission_profile(&self) {        let Ok(_refresh_guard) = self.managed_network_proxy_refresh_lock.acquire().await else {            error!("managed network proxy refresh semaphore closed");            return;        };        let session_configuration = {            let state = self.state.lock().await;            state.session_configuration.clone()        };        let Some(spec) = session_configuration            .original_config_do_not_use            .permissions            .network            .as_ref()            .cloned()        else {            self.services.network_proxy.store(None);            return;        };        let spec = match spec            .recompute_for_permission_profile(&session_configuration.permission_profile())        {            Ok(spec) => spec,            Err(err) => {                warn!("failed to rebuild managed network proxy policy for sandbox change: {err}");                return;            }        };        let current_exec_policy = self.services.exec_policy.current();        let spec = match spec.with_exec_policy_network_rules(current_exec_policy.as_ref()) {            Ok(spec) => spec,            Err(err) => {                warn!(                    "failed to apply execpolicy network rules while refreshing managed network proxy: {err}"                );                spec            }        };        if let Some(started_proxy) = self.services.network_proxy.load_full() {            if let Err(err) = spec.apply_to_started_proxy(started_proxy.as_ref()).await {                warn!("failed to refresh managed network proxy for sandbox change: {err}");            }            return;        }        match Self::start_managed_network_proxy(            &spec,            current_exec_policy.as_ref(),            &session_configuration.permission_profile(),            /*network_policy_decider*/ None,            self.services                .managed_network_requirements_configured                .then(|| {                    build_blocked_request_observer(Arc::clone(&self.services.network_approval))                }),            self.services.managed_network_requirements_configured,            self.services.network_proxy_audit_metadata.clone(),        )        .await        {            Ok((started_proxy, _session_network_proxy)) => {                self.services                    .network_proxy                    .store(Some(Arc::new(started_proxy)));            }            Err(err) => {                warn!("failed to start managed network proxy for sandbox change: {err}");            }        }    }    #[cfg(test)]    pub(crate) async fn codex_home(&self) -> AbsolutePathBuf {        let state = self.state.lock().await;        state.session_configuration.codex_home().clone()    }    pub(crate) fn subscribe_out_of_band_elicitation_pause_state(&self) -> watch::Receiver<bool> {        self.out_of_band_elicitation_paused.subscribe()    }    pub(crate) fn set_out_of_band_elicitation_pause_state(&self, paused: bool) {        self.out_of_band_elicitation_paused.send_replace(paused);    }    pub(crate) fn get_tx_event(&self) -> Sender<Event> {        self.tx_event.clone()    }    pub(crate) fn state_db(&self) -> Option<state_db::StateDbHandle> {        self.services.state_db.clone()    }    pub(crate) fn live_thread_for_persistence(        &self,        operation: &str,    ) -> anyhow::Result<&LiveThread> {        self.live_thread()            .ok_or_else(|| anyhow::anyhow!("Session persistence is disabled; cannot {operation}."))    }    pub(crate) fn live_thread(&self) -> Option<&LiveThread> {        self.services.live_thread.as_ref()    }    /// Flush rollout writes and return the final durability-barrier result.    #[instrument(name = "session.flush_rollout", level = "trace", skip_all)]    pub(crate) async fn flush_rollout(&self) -> std::io::Result<()> {        if let Some(live_thread) = self.live_thread() {            live_thread.flush().await.map_err(std::io::Error::other)        } else {            Ok(())        }    }    pub(crate) async fn try_ensure_rollout_materialized(&self) -> std::io::Result<()> {        if let Some(live_thread) = self.live_thread() {            live_thread.persist().await.map_err(std::io::Error::other)?;        }        Ok(())    }    pub(crate) async fn ensure_rollout_materialized(&self) {        if let Err(e) = self.try_ensure_rollout_materialized().await {            warn!("failed to materialize thread persistence: {e}");        }    }    fn next_internal_sub_id(&self) -> String {        let id = self            .next_internal_sub_id            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);        format!("auto-compact-{id}")    }    pub(crate) async fn route_realtime_text_input(self: &Arc<Self>, text: String) {        handlers::user_input_or_turn_inner(            self,            self.next_internal_sub_id(),            Op::UserInput {                items: vec![UserInput::Text {                    text,                    text_elements: Vec::new(),                }],                final_output_json_schema: None,                responsesapi_client_metadata: None,                additional_context: Default::default(),                thread_settings: Default::default(),            },            /*client_user_message_id*/ None,        )        .await;    }    pub(crate) async fn get_total_token_usage(&self) -> i64 {        let state = self.state.lock().await;        state.get_total_token_usage(state.server_reasoning_included())    }    pub(crate) async fn auto_compact_window_snapshot(&self) -> AutoCompactWindowSnapshot {        let state = self.state.lock().await;        state.auto_compact_window_snapshot()    }    pub(crate) async fn estimated_tokens_after_last_model_generated_item(&self) -> i64 {        let state = self.state.lock().await;        state            .history            .estimated_tokens_after_last_model_generated_item()    }    pub(crate) async fn total_token_usage(&self) -> Option<TokenUsage> {        let state = self.state.lock().await;        state.token_info().map(|info| info.total_token_usage)    }    /// Returns the complete token usage snapshot currently cached for this session.    ///    /// Resume and fork reconstruction seed this state from the last persisted rollout    /// `TokenCount` event. Callers that need to replay restored usage to a client    /// should use this accessor instead of `total_token_usage`, because the app-server    /// notification includes both total and last-turn usage.    pub(crate) async fn token_usage_info(&self) -> Option<TokenUsageInfo> {        let state = self.state.lock().await;        state.token_info()    }    pub(crate) async fn get_estimated_token_count(        &self,        turn_context: &TurnContext,    ) -> Option<i64> {        let state = self.state.lock().await;        state.history.estimate_token_count(turn_context)    }    pub(crate) async fn get_base_instructions(&self) -> BaseInstructions {        let state = self.state.lock().await;        BaseInstructions {            text: state.session_configuration.base_instructions.clone(),        }    }    // Merges connector IDs into the session-level explicit connector selection.    pub(crate) async fn merge_connector_selection(        &self,        connector_ids: HashSet<String>,    ) -> HashSet<String> {        let mut state = self.state.lock().await;        state.merge_connector_selection(connector_ids)    }    // Returns the connector IDs currently selected for this session.    pub(crate) async fn get_connector_selection(&self) -> HashSet<String> {        let state = self.state.lock().await;        state.get_connector_selection()    }    // Clears connector IDs that were accumulated for explicit selection.    pub(crate) async fn clear_connector_selection(&self) {        let mut state = self.state.lock().await;        state.clear_connector_selection();    }    async fn record_initial_history(&self, conversation_history: InitialHistory) {        let is_subagent = {            let state = self.state.lock().await;            state                .session_configuration                .session_source                .is_non_root_agent()        };        let has_prior_user_turns = initial_history_has_prior_user_turns(&conversation_history);        {            let mut state = self.state.lock().await;            state.set_next_turn_is_first(!has_prior_user_turns);        }        match conversation_history {            InitialHistory::New | InitialHistory::Cleared => {                // Defer initial context insertion until the first real turn starts so                // turn/start overrides can be merged before we write model-visible context.                self.set_previous_turn_settings(/*previous_turn_settings*/ None)                    .await;            }            InitialHistory::Resumed(resumed_history) => {                let turn_context = self.new_default_turn().await;                let rollout_items = resumed_history.history;                let previous_turn_settings = self                    .apply_rollout_reconstruction(&turn_context, &rollout_items)                    .await;                // If resuming, warn when the last recorded model differs from the current one.                let curr: &str = turn_context.model_info.slug.as_str();                if let Some(prev) = previous_turn_settings                    .as_ref()                    .map(|settings| settings.model.as_str())                    .filter(|model| *model != curr)                {                    warn!("resuming session with different model: previous={prev}, current={curr}");                    self.send_event(                        &turn_context,                        EventMsg::Warning(WarningEvent {                            message: format!(                                "This session was recorded with model `{prev}` but is resuming with `{curr}`. \                         Consider switching back to `{prev}` as it may affect Codex performance."                            ),                        }),                    )                    .await;                }                // Seed usage info from the recorded rollout so UIs can show token counts                // immediately on resume/fork.                if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {                    let mut state = self.state.lock().await;                    state.set_token_info(Some(info));                }                // Defer seeding the session's initial context until the first turn starts so                // turn/start overrides can be merged before we write to the rollout.                if !is_subagent {                    let _ = self.flush_rollout().await;                }            }            InitialHistory::Forked(rollout_items) => {                let turn_context = self.new_default_turn().await;                self.apply_rollout_reconstruction(&turn_context, &rollout_items)                    .await;                // Seed usage info from the recorded rollout so UIs can show token counts                // immediately on resume/fork.                if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {                    let mut state = self.state.lock().await;                    state.set_token_info(Some(info));                }                // If persisting, persist all rollout items as-is (the store filters).                if !rollout_items.is_empty() {                    self.persist_rollout_items(&rollout_items).await;                }                // Forked threads should remain file-backed immediately after startup.                self.ensure_rollout_materialized().await;                // Flush after seeding history and any persisted rollout copy.                if !is_subagent {                    let _ = self.flush_rollout().await;                }            }        }    }    #[instrument(        level = "trace",        skip_all,        fields(            thread_id = %self.thread_id(),            rollout_item_count = rollout_items.len()        )    )]    async fn apply_rollout_reconstruction(        &self,        turn_context: &TurnContext,        rollout_items: &[RolloutItem],    ) -> Option<PreviousTurnSettings> {        let rollout_reconstruction::RolloutReconstruction {            mut history,            previous_turn_settings,            reference_context_item,            window_id,        } = self            .reconstruct_history_from_rollout(turn_context, rollout_items)            .await;        if turn_context.features.enabled(Feature::ResizeAllImages) {            // Keep the recorded rollout unchanged. Prepare its reconstructed history before            // installing it, so legacy images are processed once for this resume or fork and            // will be processed again if the rollout is reconstructed in a future session.            // This meets image resizing requirements without modifying persisted rollouts.            prepare_response_items(&mut history);        }        {            let mut state = self.state.lock().await;            state.replace_history(history, reference_context_item);            state.set_auto_compact_window_id(window_id);            state.set_previous_turn_settings(previous_turn_settings.clone());        }        let prefix_tokens = if matches!(            turn_context.config.model_auto_compact_token_limit_scope,            AutoCompactTokenLimitScope::BodyAfterPrefix        ) {            let history = self.clone_history().await;            let base_instructions = self.get_base_instructions().await;            history.estimate_token_count_with_base_instructions(&base_instructions)        } else {            None        };        if let Some(prefix_tokens) = prefix_tokens {            self.set_auto_compact_window_estimated_prefill_for_scope(turn_context, prefix_tokens)                .await;        }        previous_turn_settings    }    async fn set_auto_compact_window_estimated_prefill_for_scope(        &self,        turn_context: &TurnContext,        tokens: i64,    ) {        if !matches!(            turn_context.config.model_auto_compact_token_limit_scope,            AutoCompactTokenLimitScope::BodyAfterPrefix        ) {            return;        }        let mut state = self.state.lock().await;        state.set_auto_compact_window_estimated_prefill(tokens);    }    fn last_token_info_from_rollout(rollout_items: &[RolloutItem]) -> Option<TokenUsageInfo> {        rollout_items.iter().rev().find_map(|item| match item {            RolloutItem::EventMsg(EventMsg::TokenCount(ev)) => ev.info.clone(),            _ => None,        })    }    async fn previous_turn_settings(&self) -> Option<PreviousTurnSettings> {        let state = self.state.lock().await;        state.previous_turn_settings()    }    pub(crate) async fn set_previous_turn_settings(        &self,        previous_turn_settings: Option<PreviousTurnSettings>,    ) {        let mut state = self.state.lock().await;        state.set_previous_turn_settings(previous_turn_settings);    }    pub(crate) async fn update_settings(        &self,        updates: SessionSettingsUpdate,    ) -> ConstraintResult<()> {        let notify_config_contributors = !self.services.extensions.config_contributors().is_empty();        let (previous_config, new_config, permission_profile_changed) = {            let mut state = self.state.lock().await;            let updated = match state.session_configuration.apply(&updates) {                Ok(updated) => updated,                Err(err) => {                    warn!("rejected session settings update: {err}");                    return Err(err);                }            };            let previous_config = notify_config_contributors                .then(|| Self::build_effective_session_config(&state.session_configuration));            let new_config =                notify_config_contributors.then(|| Self::build_effective_session_config(&updated));            let previous_permission_profile = state.session_configuration.permission_profile();            let updated_permission_profile = updated.permission_profile();            let permission_profile_changed =                previous_permission_profile != updated_permission_profile;            if updates.environments.is_some() {                self.services                    .turn_environments                    .update_selections(updated.environment_selections());            }            state.session_configuration = updated;            (previous_config, new_config, permission_profile_changed)        };        self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref());        if permission_profile_changed {            self.refresh_managed_network_proxy_for_current_permission_profile()                .await;        }        Ok(())    }    pub(crate) async fn preview_settings(        &self,        updates: &SessionSettingsUpdate,    ) -> ConstraintResult<ThreadConfigSnapshot> {        let state = self.state.lock().await;        state            .session_configuration            .apply(updates)            .map(|configuration| configuration.thread_config_snapshot())    }    pub(crate) async fn set_session_startup_prewarm(        &self,        startup_prewarm: SessionStartupPrewarmHandle,    ) {        let mut state = self.state.lock().await;        state.set_session_startup_prewarm(startup_prewarm);    }    pub(crate) async fn take_session_startup_prewarm(&self) -> Option<SessionStartupPrewarmHandle> {        let mut state = self.state.lock().await;        state.take_session_startup_prewarm()    }    pub(crate) async fn get_config(&self) -> std::sync::Arc<Config> {        let state = self.state.lock().await;        state            .session_configuration            .original_config_do_not_use            .clone()    }    pub(crate) async fn user_instructions(&self) -> Option<codex_extension_api::UserInstructions> {        let state = self.state.lock().await;        state            .session_configuration            .loaded_agents_md            .as_ref()            .and_then(LoadedAgentsMd::user_instructions)            .cloned()    }    pub(crate) async fn provider(&self) -> ModelProviderInfo {        let state = self.state.lock().await;        state.session_configuration.provider.clone()    }    pub(crate) async fn refresh_runtime_config(&self, next_config: Config) {        // Refresh only the user layer from the incoming snapshot. Preserve thread-local        // layers such as request/session overrides that were present when this session        // was created.        let notify_config_contributors = !self.services.extensions.config_contributors().is_empty();        let (previous_config, new_config, config) = {            let mut state = self.state.lock().await;            let previous_config = notify_config_contributors                .then(|| Self::build_effective_session_config(&state.session_configuration));            let mut config = (*state.session_configuration.original_config_do_not_use).clone();            config.config_layer_stack = config                .config_layer_stack                .with_user_layer_from(&next_config.config_layer_stack);            config.tool_suggest =                resolve_tool_suggest_config_from_layer_stack(&config.config_layer_stack);            let config = Arc::new(config);            state.session_configuration.original_config_do_not_use = Arc::clone(&config);            let new_config = notify_config_contributors                .then(|| Self::build_effective_session_config(&state.session_configuration));            (previous_config, new_config, config)        };        self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref());        self.services.skills_manager.clear_cache();        self.services.plugins_manager.clear_cache();        let environments = self.services.turn_environments.snapshot().await;        let hooks = build_hooks_for_config(            config.as_ref(),            self.services.plugins_manager.as_ref(),            environments.single_local_environment(),        )        .await;        let state = self.state.lock().await;        // A newer refresh may have updated the config while this hook build was in flight.        // Only publish hooks derived from the current config snapshot.        if Arc::ptr_eq(            &state.session_configuration.original_config_do_not_use,            &config,        ) {            self.services.hooks.store(Arc::new(hooks));        }    }    fn emit_config_changed_contributors(        &self,        previous_config: Option<&Config>,        new_config: Option<&Config>,    ) {        let (Some(previous_config), Some(new_config)) = (previous_config, new_config) else {            return;        };        if previous_config == new_config {            return;        }        for contributor in self.services.extensions.config_contributors() {            contributor.on_config_changed(                &self.services.session_extension_data,                &self.services.thread_extension_data,                previous_config,                new_config,            );        }    }    pub(crate) async fn reload_user_config_layer(&self) {        // Refresh layer-backed runtime state for an existing session, including enabled plugin,        // skill, and hook state. Derived config fields such as feature gates and legacy notify        // settings remain session-static.        //        // Prefer `refresh_runtime_config()` when the host can already provide a materialized        // config snapshot. This file-based path exists for legacy local reload flows.        let config_toml_paths = {            let state = self.state.lock().await;            let config = &state.session_configuration.original_config_do_not_use;            let user_config_paths = config                .config_layer_stack                .get_user_layers(                    ConfigLayerStackOrdering::LowestPrecedenceFirst,                    /*include_disabled*/ true,                )                .into_iter()                .filter_map(|layer| match &layer.name {                    ConfigLayerSource::User { file, .. } => Some(file.clone()),                    _ => None,                })                .collect::<Vec<_>>();            if user_config_paths.is_empty() {                vec![                    state                        .session_configuration                        .codex_home                        .join(CONFIG_TOML_FILE),                ]            } else {                user_config_paths            }        };        let mut reloaded_user_configs = Vec::with_capacity(config_toml_paths.len());        for config_toml_path in config_toml_paths {            let user_config = match std::fs::read_to_string(&config_toml_path) {                Ok(contents) => match toml::from_str::<toml::Value>(&contents) {                    Ok(config) => config,                    Err(err) => {                        warn!("failed to parse user config while reloading layer: {err}");                        return;                    }                },                Err(err) if err.kind() == std::io::ErrorKind::NotFound => {                    toml::Value::Table(Default::default())                }                Err(err) => {                    warn!("failed to read user config while reloading layer: {err}");                    return;                }            };            reloaded_user_configs.push((config_toml_path, user_config));        }        let next_config = {            let state = self.state.lock().await;            let mut config = (*state.session_configuration.original_config_do_not_use).clone();            for (config_toml_path, user_config) in reloaded_user_configs {                config.config_layer_stack = config                    .config_layer_stack                    .with_user_config(&config_toml_path, user_config);            }            config.tool_suggest =                resolve_tool_suggest_config_from_layer_stack(&config.config_layer_stack);            config        };        self.refresh_runtime_config(next_config).await;    }    async fn build_settings_update_items(        &self,        reference_context_item: Option<&TurnContextItem>,        current_context: &TurnContext,    ) -> Vec<ResponseItem> {        // TODO: Make context updates a pure diff of persisted previous/current TurnContextItem        // state so replay/backtracking is deterministic. Runtime inputs that affect model-visible        // context (shell, exec policy, feature gates, previous-turn bridge) should be persisted        // state or explicit non-state replay events.        let previous_turn_settings = {            let state = self.state.lock().await;            state.previous_turn_settings()        };        let shell = self.user_shell();        let exec_policy = self.services.exec_policy.current();        crate::context_manager::updates::build_settings_update_items(            reference_context_item,            previous_turn_settings.as_ref(),            current_context,            shell.as_ref(),            exec_policy.as_ref(),            self.features.enabled(Feature::Personality),        )    }    /// Record a terminal CodexErr before the app-server completion notification is reduced.    pub(crate) fn track_turn_codex_error(&self, turn_context: &TurnContext, error: &CodexErr) {        self.services            .analytics_events_client            .track_turn_codex_error(TurnCodexErrorFact::from_codex_err(                self.thread_id.to_string(),                turn_context.sub_id.clone(),                error,            ));    }    /// Persist the event to rollout and send it to clients.    pub(crate) async fn send_event(&self, turn_context: &TurnContext, msg: EventMsg) {        let legacy_source = msg.clone();        if let EventMsg::Error(error) = &legacy_source            && error                .codex_error_info                .as_ref()                .is_some_and(CodexErrorInfo::affects_turn_status)        {            turn_context                .terminal_error                .lock()                .await                .replace(error.message.clone());        }        self.services            .rollout_thread_trace            .record_codex_turn_event(&turn_context.sub_id, &legacy_source);        self.services            .rollout_thread_trace            .record_tool_call_event(turn_context.sub_id.clone(), &legacy_source);        let event = Event {            id: turn_context.sub_id.clone(),            msg,        };        self.send_event_raw(event).await;        self.maybe_notify_parent_of_terminal_turn(turn_context, &legacy_source)            .await;        self.maybe_mirror_event_text_to_realtime(&legacy_source)            .await;        self.maybe_clear_realtime_handoff_for_event(&legacy_source)            .await;        let show_raw_agent_reasoning = self.show_raw_agent_reasoning();        for legacy in legacy_source.as_legacy_events(show_raw_agent_reasoning) {            let legacy_event = Event {                id: turn_context.sub_id.clone(),                msg: legacy,            };            self.send_event_raw(legacy_event).await;        }    }    /// Forwards terminal turn events from spawned MultiAgentV2 children to their direct parent.    async fn maybe_notify_parent_of_terminal_turn(        &self,        turn_context: &TurnContext,        msg: &EventMsg,    ) {        if turn_context.multi_agent_version != MultiAgentVersion::V2 {            return;        }        if !matches!(msg, EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_)) {            return;        }        let SessionSource::SubAgent(SubAgentSource::ThreadSpawn {            parent_thread_id,            agent_path: Some(child_agent_path),            ..        }) = &turn_context.session_source        else {            return;        };        let status = match turn_context.terminal_error.lock().await.take() {            Some(error) => {                let status = AgentStatus::Errored(error);                self.agent_status.send_replace(status.clone());                status            }            None => {                let Some(status) = agent_status_from_event(msg) else {                    return;                };                status            }        };        if !is_final(&status) {            return;        }        self.forward_child_completion_to_parent(            turn_context,            *parent_thread_id,            child_agent_path,            status,        )        .await;    }    /// Sends the standard completion envelope from a spawned MultiAgentV2 child to its parent.    async fn forward_child_completion_to_parent(        &self,        turn_context: &TurnContext,        parent_thread_id: ThreadId,        child_agent_path: &codex_protocol::AgentPath,        status: AgentStatus,    ) {        let Some(parent_agent_path) = child_agent_path            .as_str()            .rsplit_once('/')            .and_then(|(parent, _)| codex_protocol::AgentPath::try_from(parent).ok())        else {            return;        };        let Some(message) = format_inter_agent_completion_message(            parent_agent_path.clone(),            child_agent_path.clone(),            &status,        ) else {            return;        };        // `communication` owns the message. Keep a second copy only when the        // recorder will actually need it after parent delivery succeeds.        let trace_message = self            .services            .rollout_thread_trace            .is_enabled()            .then(|| message.clone());        let communication = InterAgentCommunication::new(            child_agent_path.clone(),            parent_agent_path,            Vec::new(),            message,            /*trigger_turn*/ false,        );        if let Err(err) = self            .services            .agent_control            .send_inter_agent_communication(parent_thread_id, communication)            .await        {            debug!("failed to notify parent thread {parent_thread_id}: {err}");            return;        }        if let Some(message) = trace_message {            self.services                .rollout_thread_trace                .record_agent_result_interaction(                    turn_context.sub_id.as_str(),                    parent_thread_id,                    &AgentResultTracePayload {                        child_agent_path: child_agent_path.as_str(),                        message: &message,                        status: &status,                    },                );        }    }    async fn maybe_mirror_event_text_to_realtime(&self, msg: &EventMsg) {        let Some(text) = realtime_text_for_event(msg) else {            return;        };        if self.conversation.running_state().await.is_none() {            return;        }        if let Err(err) = self.conversation.handoff_out(text).await {            debug!("failed to mirror event text to realtime conversation: {err}");        }    }    async fn maybe_clear_realtime_handoff_for_event(&self, msg: &EventMsg) {        if !matches!(msg, EventMsg::TurnComplete(_)) {            return;        }        if let Err(err) = self.conversation.handoff_complete().await {            debug!("failed to finalize realtime handoff output: {err}");        }        self.conversation.clear_active_handoff().await;    }    pub(crate) async fn send_event_raw(&self, event: Event) {        // Persist the event into rollout storage (the store filters as needed).        let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())];        self.persist_rollout_items(&rollout_items).await;        self.services            .rollout_thread_trace            .record_protocol_event(&event.msg);        self.deliver_event_raw(event).await;    }    async fn deliver_event_raw(&self, event: Event) {        // Record the last known agent status.        if let Some(status) = agent_status_from_event(&event.msg) {            self.agent_status.send_replace(status);        }        if let Err(e) = self.tx_event.send(event).await {            debug!("dropping event because channel is closed: {e}");        }    }    pub(crate) async fn emit_turn_item_started(&self, turn_context: &TurnContext, item: &TurnItem) {        self.send_event(            turn_context,            EventMsg::ItemStarted(ItemStartedEvent {                thread_id: self.thread_id,                turn_id: turn_context.sub_id.clone(),                item: item.clone(),                started_at_ms: now_unix_timestamp_ms(),            }),        )        .await;    }    pub(crate) async fn emit_turn_item_completed(        &self,        turn_context: &TurnContext,        item: TurnItem,    ) {        record_turn_ttfm_metric(turn_context, &item).await;        self.send_event(            turn_context,            EventMsg::ItemCompleted(ItemCompletedEvent {                thread_id: self.thread_id,                turn_id: turn_context.sub_id.clone(),                item,                completed_at_ms: now_unix_timestamp_ms(),            }),        )        .await;    }    /// Adds an execpolicy amendment to both the in-memory and on-disk policies so future    /// commands can use the newly approved prefix.    pub(crate) async fn persist_execpolicy_amendment(        &self,        amendment: &ExecPolicyAmendment,    ) -> Result<(), ExecPolicyUpdateError> {        let codex_home = self            .state            .lock()            .await            .session_configuration            .codex_home()            .clone();        self.services            .exec_policy            .append_amendment_and_update(&codex_home, amendment)            .await?;        Ok(())    }    pub(crate) async fn turn_context_for_sub_id(&self, sub_id: &str) -> Option<Arc<TurnContext>> {        let active = self.active_turn.lock().await;        active            .as_ref()            .and_then(|turn| turn.task.as_ref())            .filter(|task| task.turn_context.sub_id == sub_id)            .map(|task| Arc::clone(&task.turn_context))    }    async fn active_turn_context_and_cancellation_token(        &self,    ) -> Option<(Arc<TurnContext>, CancellationToken)> {        let active = self.active_turn.lock().await;        let task = active.as_ref()?.task.as_ref()?;        Some((            Arc::clone(&task.turn_context),            task.cancellation_token.child_token(),        ))    }    pub(crate) async fn record_execpolicy_amendment_message(        &self,        sub_id: &str,        amendment: &ExecPolicyAmendment,    ) {        let Some(prefixes) = format_allow_prefixes(vec![amendment.command.clone()]) else {            warn!("execpolicy amendment for {sub_id} had no command prefix");            return;        };        let message: ResponseItem =            ContextualUserFragment::into(ApprovedCommandPrefixSaved::new(prefixes));        let turn_context = self.turn_context_for_sub_id(sub_id).await;        self.inject_no_new_turn(vec![message], turn_context.as_deref())            .await;    }    pub(crate) async fn persist_network_policy_amendment(        &self,        amendment: &NetworkPolicyAmendment,        network_approval_context: &NetworkApprovalContext,    ) -> anyhow::Result<()> {        let _refresh_guard = self            .managed_network_proxy_refresh_lock            .acquire()            .await            .map_err(|_| anyhow::anyhow!("managed network proxy refresh semaphore closed"))?;        let host =            Self::validated_network_policy_amendment_host(amendment, network_approval_context)?;        let codex_home = self            .state            .lock()            .await            .session_configuration            .codex_home()            .clone();        let execpolicy_amendment =            execpolicy_network_rule_amendment(amendment, network_approval_context, &host);        if let Some(started_network_proxy) = self.services.network_proxy.load_full() {            let proxy = started_network_proxy.proxy();            match amendment.action {                NetworkPolicyRuleAction::Allow => proxy                    .add_allowed_domain(&host)                    .await                    .map_err(|err| anyhow::anyhow!("failed to update runtime allowlist: {err}"))?,                NetworkPolicyRuleAction::Deny => proxy                    .add_denied_domain(&host)                    .await                    .map_err(|err| anyhow::anyhow!("failed to update runtime denylist: {err}"))?,            }        }        self.services            .exec_policy            .append_network_rule_and_update(                &codex_home,                &host,                execpolicy_amendment.protocol,                execpolicy_amendment.decision,                Some(execpolicy_amendment.justification),            )            .await            .map_err(|err| {                anyhow::anyhow!("failed to persist network policy amendment to execpolicy: {err}")            })?;        Ok(())    }    fn validated_network_policy_amendment_host(        amendment: &NetworkPolicyAmendment,        network_approval_context: &NetworkApprovalContext,    ) -> anyhow::Result<String> {        let approved_host = normalize_host(&network_approval_context.host);        let amendment_host = normalize_host(&amendment.host);        if amendment_host != approved_host {            return Err(anyhow::anyhow!(                "network policy amendment host '{}' does not match approved host '{}'",                amendment.host,                network_approval_context.host            ));        }        Ok(approved_host)    }    pub(crate) async fn record_network_policy_amendment_message(        &self,        sub_id: &str,        amendment: &NetworkPolicyAmendment,    ) {        let message: ResponseItem = ContextualUserFragment::into(NetworkRuleSaved::new(amendment));        let turn_context = self.turn_context_for_sub_id(sub_id).await;        self.inject_no_new_turn(vec![message], turn_context.as_deref())            .await;    }    /// Emit an exec approval request event and await the user's decision.    ///    /// The request is keyed by `call_id` + `approval_id` so matching responses    /// are delivered to the correct in-flight turn. If the pending approval is    /// cleared before a response arrives, treat it as an abort so interrupted    /// turns do not continue on a synthetic denial.    ///    /// Note that if `available_decisions` is `None`, then the other fields will    /// be used to derive the available decisions via    /// [ExecApprovalRequestEvent::default_available_decisions].    #[allow(clippy::too_many_arguments)]    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn request_command_approval(        &self,        turn_context: &TurnContext,        call_id: String,        approval_id: Option<String>,        command: Vec<String>,        cwd: AbsolutePathBuf,        reason: Option<String>,        network_approval_context: Option<NetworkApprovalContext>,        proposed_execpolicy_amendment: Option<ExecPolicyAmendment>,        additional_permissions: Option<AdditionalPermissionProfile>,        available_decisions: Option<Vec<ReviewDecision>>,    ) -> ReviewDecision {        //  command-level approvals use `call_id`.        // `approval_id` is only present for subcommand callbacks (execve intercept)        let effective_approval_id = approval_id.clone().unwrap_or_else(|| call_id.clone());        // Add the tx_approve callback to the map before sending the request.        let (tx_approve, rx_approve) = oneshot::channel();        let prev_entry = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    ts.insert_pending_approval(effective_approval_id.clone(), tx_approve)                }                None => None,            }        };        if prev_entry.is_some() {            warn!("Overwriting existing pending approval for call_id: {effective_approval_id}");        }        let parsed_cmd = parse_command(&command);        let proposed_network_policy_amendments = network_approval_context.as_ref().map(|context| {            vec![                NetworkPolicyAmendment {                    host: context.host.clone(),                    action: NetworkPolicyRuleAction::Allow,                },                NetworkPolicyAmendment {                    host: context.host.clone(),                    action: NetworkPolicyRuleAction::Deny,                },            ]        });        let available_decisions = available_decisions.unwrap_or_else(|| {            ExecApprovalRequestEvent::default_available_decisions(                network_approval_context.as_ref(),                proposed_execpolicy_amendment.as_ref(),                proposed_network_policy_amendments.as_deref(),                additional_permissions.as_ref(),            )        });        let event = EventMsg::ExecApprovalRequest(ExecApprovalRequestEvent {            call_id,            approval_id,            turn_id: turn_context.sub_id.clone(),            started_at_ms: now_unix_timestamp_ms(),            command,            cwd,            reason,            network_approval_context,            proposed_execpolicy_amendment,            proposed_network_policy_amendments,            additional_permissions,            available_decisions: Some(available_decisions),            parsed_cmd,        });        self.send_event(turn_context, event).await;        rx_approve.await.unwrap_or(ReviewDecision::Abort)    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn request_patch_approval(        &self,        turn_context: &TurnContext,        call_id: String,        changes: HashMap<PathBuf, FileChange>,        reason: Option<String>,        grant_root: Option<PathBuf>,    ) -> oneshot::Receiver<ReviewDecision> {        // Add the tx_approve callback to the map before sending the request.        let (tx_approve, rx_approve) = oneshot::channel();        let approval_id = call_id.clone();        let prev_entry = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    ts.insert_pending_approval(approval_id.clone(), tx_approve)                }                None => None,            }        };        if prev_entry.is_some() {            warn!("Overwriting existing pending approval for call_id: {approval_id}");        }        let event = EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent {            call_id,            turn_id: turn_context.sub_id.clone(),            started_at_ms: now_unix_timestamp_ms(),            changes,            reason,            grant_root,        });        self.send_event(turn_context, event).await;        rx_approve    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub(crate) async fn request_permissions_for_environment(        self: &Arc<Self>,        turn_context: &Arc<TurnContext>,        call_id: String,        args: RequestPermissionsArgs,        environment: TurnEnvironmentSelection,        cancellation_token: CancellationToken,    ) -> Option<RequestPermissionsResponse> {        match turn_context.as_ref().approval_policy.value() {            AskForApproval::Never => {                return Some(RequestPermissionsResponse {                    permissions: RequestPermissionProfile::default(),                    scope: PermissionGrantScope::Turn,                    strict_auto_review: false,                });            }            AskForApproval::Granular(granular_config)                if !granular_config.allows_request_permissions() =>            {                return Some(RequestPermissionsResponse {                    permissions: RequestPermissionProfile::default(),                    scope: PermissionGrantScope::Turn,                    strict_auto_review: false,                });            }            AskForApproval::OnFailure            | AskForApproval::OnRequest            | AskForApproval::UnlessTrusted            | AskForApproval::Granular(_) => {}        }        let requested_permissions = args.permissions;        // TODO(anp): Migrate request_permissions to support paths from foreign environments.        let Ok(native_environment_cwd) = environment.cwd.to_abs_path() else {            warn!(                cwd = %environment.cwd,                "request_permissions requires a cwd native to the Codex host"            );            return Some(RequestPermissionsResponse {                permissions: RequestPermissionProfile::default(),                scope: PermissionGrantScope::Turn,                strict_auto_review: false,            });        };        if crate::guardian::routes_approval_to_guardian(turn_context.as_ref()) {            let originating_turn_state = {                let active = self.active_turn.lock().await;                active.as_ref().map(|active| Arc::clone(&active.turn_state))            };            let review_id = crate::guardian::new_guardian_review_id();            let session = Arc::clone(self);            let turn = Arc::clone(turn_context);            let request = crate::guardian::GuardianApprovalRequest::RequestPermissions {                id: call_id,                turn_id: turn_context.sub_id.clone(),                reason: args.reason,                permissions: requested_permissions.clone(),            };            let review_rx = crate::guardian::spawn_approval_request_review(                session,                turn,                review_id,                request,                /*retry_reason*/ None,                codex_analytics::GuardianApprovalRequestSource::MainTurn,                cancellation_token.clone(),            );            let decision = tokio::select! {                biased;                _ = cancellation_token.cancelled() => return None,                decision = review_rx => decision.unwrap_or(ReviewDecision::Denied),            };            let response = match decision {                ReviewDecision::Approved | ReviewDecision::ApprovedExecpolicyAmendment { .. } => {                    RequestPermissionsResponse {                        permissions: requested_permissions.clone(),                        scope: PermissionGrantScope::Turn,                        strict_auto_review: false,                    }                }                ReviewDecision::ApprovedForSession => RequestPermissionsResponse {                    permissions: requested_permissions.clone(),                    scope: PermissionGrantScope::Session,                    strict_auto_review: false,                },                ReviewDecision::NetworkPolicyAmendment {                    network_policy_amendment,                } => match network_policy_amendment.action {                    NetworkPolicyRuleAction::Allow => RequestPermissionsResponse {                        permissions: requested_permissions.clone(),                        scope: PermissionGrantScope::Turn,                        strict_auto_review: false,                    },                    NetworkPolicyRuleAction::Deny => RequestPermissionsResponse {                        permissions: RequestPermissionProfile::default(),                        scope: PermissionGrantScope::Turn,                        strict_auto_review: false,                    },                },                ReviewDecision::Abort | ReviewDecision::Denied | ReviewDecision::TimedOut => {                    RequestPermissionsResponse {                        permissions: RequestPermissionProfile::default(),                        scope: PermissionGrantScope::Turn,                        strict_auto_review: false,                    }                }            };            let response = Self::normalize_request_permissions_response(                requested_permissions,                response,                native_environment_cwd.as_path(),            );            self.record_granted_request_permissions_for_turn(                &response,                &environment.environment_id,                originating_turn_state.as_ref(),            )            .await;            return Some(response);        }        let (tx_response, rx_response) = oneshot::channel();        let prev_entry = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    ts.insert_pending_request_permissions(                        call_id.clone(),                        PendingRequestPermissions {                            tx_response,                            requested_permissions: requested_permissions.clone(),                            environment: environment.clone(),                        },                    )                }                None => None,            }        };        if prev_entry.is_some() {            warn!("Overwriting existing pending request_permissions for call_id: {call_id}");        }        let event = EventMsg::RequestPermissions(RequestPermissionsEvent {            call_id: call_id.clone(),            turn_id: turn_context.sub_id.clone(),            environment_id: Some(environment.environment_id.clone()),            started_at_ms: now_unix_timestamp_ms(),            reason: args.reason,            permissions: requested_permissions,            cwd: Some(native_environment_cwd),        });        self.send_event(turn_context.as_ref(), event).await;        tokio::select! {            biased;            _ = cancellation_token.cancelled() => {                let mut active = self.active_turn.lock().await;                if let Some(at) = active.as_mut() {                    let mut ts = at.turn_state.lock().await;                    let _ = ts.remove_pending_request_permissions(&call_id);                }                None            }            response = rx_response => response.ok(),        }    }    pub(crate) async fn request_permissions_for_cwd(        self: &Arc<Self>,        turn_context: &Arc<TurnContext>,        call_id: String,        args: RequestPermissionsArgs,        cwd: AbsolutePathBuf,        cancellation_token: CancellationToken,    ) -> Option<RequestPermissionsResponse> {        let turn_environment = match args.environment_id.as_deref() {            Some(environment_id) => turn_context                .environments                .turn_environments                .iter()                .find(|environment| environment.environment_id == environment_id),            None => turn_context.environments.primary(),        };        let Some(turn_environment) = turn_environment else {            return Some(RequestPermissionsResponse {                permissions: RequestPermissionProfile::default(),                scope: PermissionGrantScope::Turn,                strict_auto_review: false,            });        };        let mut environment = turn_environment.selection();        environment.cwd = PathUri::from_abs_path(&cwd);        self.request_permissions_for_environment(            turn_context,            call_id,            args,            environment,            cancellation_token,        )        .await    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn request_user_input(        &self,        turn_context: &TurnContext,        call_id: String,        args: RequestUserInputArgs,    ) -> Option<RequestUserInputResponse> {        let sub_id = turn_context.sub_id.clone();        let (tx_response, rx_response) = oneshot::channel();        let event_id = sub_id.clone();        let prev_entry = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    ts.insert_pending_user_input(sub_id, tx_response)                }                None => None,            }        };        if prev_entry.is_some() {            warn!("Overwriting existing pending user input for sub_id: {event_id}");        }        let event = EventMsg::RequestUserInput(RequestUserInputEvent {            call_id,            turn_id: turn_context.sub_id.clone(),            questions: args.questions,            auto_resolution_ms: args.auto_resolution_ms,        });        turn_context            .turn_metadata_state            .mark_user_input_requested_during_turn();        self.send_event(turn_context, event).await;        rx_response.await.ok()    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn notify_user_input_response(        &self,        sub_id: &str,        response: RequestUserInputResponse,    ) {        let entry = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    ts.remove_pending_user_input(sub_id)                }                None => None,            }        };        match entry {            Some(tx_response) => {                tx_response.send(response).ok();            }            None => {                warn!("No pending user input found for sub_id: {sub_id}");            }        }    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn notify_request_permissions_response(        &self,        call_id: &str,        response: RequestPermissionsResponse,    ) {        let (entry, originating_turn_state) = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    let entry = ts.remove_pending_request_permissions(call_id);                    let originating_turn_state = entry.as_ref().map(|_| Arc::clone(&at.turn_state));                    (entry, originating_turn_state)                }                None => (None, None),            }        };        match entry {            Some(entry) => {                // TODO(anp): Migrate request_permissions to support paths from foreign environments.                let response = match entry.environment.cwd.to_abs_path() {                    Ok(native_environment_cwd) => Self::normalize_request_permissions_response(                        entry.requested_permissions,                        response,                        native_environment_cwd.as_path(),                    ),                    Err(err) => {                        warn!(                            cwd = %entry.environment.cwd,                            %err,                            "request_permissions requires a cwd native to the Codex host"                        );                        RequestPermissionsResponse {                            permissions: RequestPermissionProfile::default(),                            scope: PermissionGrantScope::Turn,                            strict_auto_review: false,                        }                    }                };                self.record_granted_request_permissions_for_turn(                    &response,                    &entry.environment.environment_id,                    originating_turn_state.as_ref(),                )                .await;                entry.tx_response.send(response).ok();            }            None => {                warn!("No pending request_permissions found for call_id: {call_id}");            }        }    }    fn normalize_request_permissions_response(        requested_permissions: RequestPermissionProfile,        response: RequestPermissionsResponse,        cwd: &Path,    ) -> RequestPermissionsResponse {        if response.strict_auto_review && matches!(response.scope, PermissionGrantScope::Session) {            return RequestPermissionsResponse {                permissions: RequestPermissionProfile::default(),                scope: PermissionGrantScope::Turn,                strict_auto_review: false,            };        }        if response.permissions.is_empty() {            return response;        }        RequestPermissionsResponse {            permissions: intersect_permission_profiles(                requested_permissions.into(),                response.permissions.into(),                cwd,            )            .into(),            scope: response.scope,            strict_auto_review: response.strict_auto_review,        }    }    async fn record_granted_request_permissions_for_turn(        &self,        response: &RequestPermissionsResponse,        environment_id: &str,        originating_turn_state: Option<&Arc<Mutex<crate::state::TurnState>>>,    ) {        if response.permissions.is_empty() {            return;        }        match response.scope {            PermissionGrantScope::Turn => {                if let Some(turn_state) = originating_turn_state {                    let mut ts = turn_state.lock().await;                    let permissions: AdditionalPermissionProfile =                        response.permissions.clone().into();                    ts.record_granted_permissions(environment_id, permissions);                    if response.strict_auto_review {                        ts.enable_strict_auto_review();                    }                }            }            PermissionGrantScope::Session => {                let mut state = self.state.lock().await;                state.record_granted_permissions(                    environment_id,                    response.permissions.clone().into(),                );            }        }    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn reads must stay consistent with the matching turn state"    )]    pub(crate) async fn granted_turn_permissions(        &self,        environment_id: &str,    ) -> Option<AdditionalPermissionProfile> {        let active = self.active_turn.lock().await;        let active = active.as_ref()?;        let ts = active.turn_state.lock().await;        ts.granted_permissions(environment_id)    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn reads must stay consistent with the matching turn state"    )]    pub(crate) async fn strict_auto_review_enabled_for_turn(&self) -> bool {        let active = self.active_turn.lock().await;        let Some(active) = active.as_ref() else {            return false;        };        let ts = active.turn_state.lock().await;        ts.strict_auto_review_enabled()    }    pub(crate) async fn granted_session_permissions(        &self,        environment_id: &str,    ) -> Option<AdditionalPermissionProfile> {        let state = self.state.lock().await;        state.granted_permissions(environment_id)    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn notify_dynamic_tool_response(&self, call_id: &str, response: DynamicToolResponse) {        let entry = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    ts.remove_pending_dynamic_tool(call_id)                }                None => None,            }        };        match entry {            Some(tx_response) => {                tx_response.send(response).ok();            }            None => {                warn!("No pending dynamic tool call found for call_id: {call_id}");            }        }    }    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn notify_approval(&self, approval_id: &str, decision: ReviewDecision) {        let entry = {            let mut active = self.active_turn.lock().await;            match active.as_mut() {                Some(at) => {                    let mut ts = at.turn_state.lock().await;                    ts.remove_pending_approval(approval_id)                }                None => None,            }        };        match entry {            Some(tx_approve) => {                tx_approve.send(decision).ok();            }            None => {                warn!("No pending approval found for call_id: {approval_id}");            }        }    }    /// Records conversation items: append to history, persist to rollout, and    /// notify clients observing raw response items.    pub(crate) fn prepare_conversation_items_for_history<'a>(        &self,        turn_context: &TurnContext,        items: &'a [ResponseItem],    ) -> Cow<'a, [ResponseItem]> {        if !turn_context.features.enabled(Feature::ResizeAllImages) {            return Cow::Borrowed(items);        }        let mut prepared_items = items.to_vec();        prepare_response_items(&mut prepared_items);        Cow::Owned(prepared_items)    }    pub(crate) fn response_item_from_user_input(        &self,        turn_context: &TurnContext,        input: Vec<UserInput>,    ) -> ResponseItem {        let local_image_preparation = if turn_context.features.enabled(Feature::ResizeAllImages) {            LocalImagePreparation::Defer        } else {            LocalImagePreparation::Process        };        ResponseItem::from(ResponseInputItem::from_user_input(            input,            local_image_preparation,        ))    }    pub(crate) async fn record_conversation_items(        &self,        turn_context: &TurnContext,        items: &[ResponseItem],    ) {        let items = self.prepare_conversation_items_for_history(turn_context, items);        let items = items.as_ref();        {            let mut state = self.state.lock().await;            state.record_items(items.iter(), turn_context.truncation_policy);        }        self.persist_rollout_response_items(items).await;        self.send_raw_response_items(turn_context, items).await;    }    pub(crate) async fn record_inter_agent_communication(        &self,        turn_context: &TurnContext,        communication: InterAgentCommunication,    ) {        let response_item = communication.to_model_input_item();        let items = self.prepare_conversation_items_for_history(            turn_context,            std::slice::from_ref(&response_item),        );        let items = items.as_ref();        {            let mut state = self.state.lock().await;            state.record_items(items.iter(), turn_context.truncation_policy);        }        self.persist_rollout_items(&[RolloutItem::InterAgentCommunication(communication)])            .await;        self.send_raw_response_items(turn_context, items).await;    }    async fn maybe_warn_on_server_model_mismatch(        self: &Arc<Self>,        turn_context: &Arc<TurnContext>,        server_model: String,    ) -> bool {        let requested_model = turn_context.model_info.slug.clone();        let server_model_normalized = server_model.to_ascii_lowercase();        let requested_model_normalized = requested_model.to_ascii_lowercase();        if server_model_normalized == requested_model_normalized {            info!("server reported model {server_model} (matches requested model)");            return false;        }        warn!("server reported model {server_model} while requested model was {requested_model}");        let warning_message = format!(            "Your account was flagged for potentially high-risk cyber activity and this request was routed to gpt-5.2 as a fallback. To regain access to gpt-5.3-codex, apply for trusted access: {CYBER_VERIFY_URL} or learn more: {CYBER_SAFETY_URL}"        );        self.send_event(            turn_context,            EventMsg::ModelReroute(ModelRerouteEvent {                from_model: requested_model.clone(),                to_model: server_model.clone(),                reason: ModelRerouteReason::HighRiskCyberActivity,            }),        )        .await;        self.send_event(            turn_context,            EventMsg::Warning(WarningEvent {                message: warning_message.clone(),            }),        )        .await;        true    }    pub(crate) async fn emit_model_verification(        self: &Arc<Self>,        turn_context: &Arc<TurnContext>,        verifications: Vec<ModelVerification>,    ) {        self.send_event(            turn_context,            EventMsg::ModelVerification(ModelVerificationEvent { verifications }),        )        .await;    }    pub(crate) async fn emit_turn_moderation_metadata(        self: &Arc<Self>,        turn_context: &Arc<TurnContext>,        metadata: TurnModerationMetadataEvent,    ) {        self.send_event(turn_context, EventMsg::TurnModerationMetadata(metadata))            .await;    }    #[cfg(test)]    pub(crate) async fn replace_history(        &self,        items: Vec<ResponseItem>,        reference_context_item: Option<TurnContextItem>,    ) {        let mut state = self.state.lock().await;        state.replace_history(items, reference_context_item);    }    pub(crate) async fn replace_compacted_history(        &self,        items: Vec<ResponseItem>,        reference_context_item: Option<TurnContextItem>,        compacted_item: CompactedItem,    ) {        {            let mut state = self.state.lock().await;            state.replace_history(items, reference_context_item.clone());        }        self.persist_rollout_items(&[RolloutItem::Compacted(compacted_item)])            .await;        if let Some(turn_context_item) = reference_context_item {            self.persist_rollout_items(&[RolloutItem::TurnContext(turn_context_item)])                .await;        }        {            let mut state = self.state.lock().await;            state.queue_pending_session_start_source(codex_hooks::SessionStartSource::Compact);        }    }    async fn persist_rollout_response_items(&self, items: &[ResponseItem]) {        let rollout_items: Vec<RolloutItem> = items            .iter()            .cloned()            .map(RolloutItem::ResponseItem)            .collect();        self.persist_rollout_items(&rollout_items).await;    }    pub fn enabled(&self, feature: Feature) -> bool {        self.features.enabled(feature)    }    pub(crate) fn features(&self) -> ManagedFeatures {        self.features.clone()    }    pub(crate) async fn collaboration_mode(&self) -> CollaborationMode {        let state = self.state.lock().await;        state.session_configuration.collaboration_mode.clone()    }    pub(crate) fn multi_agent_version(&self) -> Option<MultiAgentVersion> {        self.multi_agent_version.get().copied()    }    pub(crate) fn set_multi_agent_version_if_unset(        &self,        multi_agent_version: MultiAgentVersion,    ) -> MultiAgentVersion {        *self.multi_agent_version.get_or_init(|| multi_agent_version)    }    pub(crate) fn resolve_multi_agent_version_for_model(        &self,        model_info: &ModelInfo,        config: &Config,    ) -> MultiAgentVersion {        if let Some(multi_agent_version) = self.multi_agent_version() {            return multi_agent_version;        }        let selected = model_info            .multi_agent_version            .unwrap_or_else(|| config.multi_agent_version_from_features());        self.set_multi_agent_version_if_unset(selected)    }    async fn send_raw_response_items(&self, turn_context: &TurnContext, items: &[ResponseItem]) {        for item in items {            self.send_event(                turn_context,                EventMsg::RawResponseItem(RawResponseItemEvent { item: item.clone() }),            )            .await;        }    }    pub(crate) async fn build_initial_context(        &self,        turn_context: &TurnContext,    ) -> Vec<ResponseItem> {        let mut developer_sections = Vec::<String>::with_capacity(8);        let mut contextual_user_sections = Vec::<String>::with_capacity(2);        let mut separate_developer_sections = Vec::<String>::new();        let (            reference_context_item,            previous_turn_settings,            collaboration_mode,            base_instructions,            session_source,            auto_compact_window_id,        ) = {            let state = self.state.lock().await;            (                state.reference_context_item(),                state.previous_turn_settings(),                state.session_configuration.collaboration_mode.clone(),                state.session_configuration.base_instructions.clone(),                state.session_configuration.session_source.clone(),                state.auto_compact_window_id(),            )        };        if let Some(model_switch_message) =            crate::context_manager::updates::build_model_instructions_update_item(                previous_turn_settings.as_ref(),                turn_context,            )        {            developer_sections.push(model_switch_message);        }        if turn_context.config.include_permissions_instructions {            developer_sections.push(                PermissionsInstructions::from_permission_profile(                    &turn_context.permission_profile,                    turn_context.approval_policy.value(),                    turn_context.config.approvals_reviewer,                    self.services.exec_policy.current().as_ref(),                    #[allow(deprecated)]                    &turn_context.cwd,                    turn_context                        .features                        .enabled(Feature::ExecPermissionApprovals),                    turn_context                        .features                        .enabled(Feature::RequestPermissionsTool),                )                .render(),            );        }        let separate_guardian_developer_message =            crate::guardian::is_guardian_reviewer_source(&session_source);        // Keep the guardian policy prompt out of the aggregated developer bundle so it        // stays isolated as its own top-level developer message for guardian subagents.        if !separate_guardian_developer_message            && let Some(developer_instructions) = turn_context.developer_instructions.as_deref()            && !developer_instructions.is_empty()        {            developer_sections.push(developer_instructions.to_string());        }        // Add developer instructions from collaboration_mode if they exist and are non-empty        if turn_context.config.include_collaboration_mode_instructions            && let Some(collab_instructions) =                CollaborationModeInstructions::from_collaboration_mode(&collaboration_mode)        {            developer_sections.push(collab_instructions.render());        }        if let Some(realtime_update) = crate::context_manager::updates::build_initial_realtime_item(            reference_context_item.as_ref(),            previous_turn_settings.as_ref(),            turn_context,        ) {            developer_sections.push(realtime_update);        }        if self.features.enabled(Feature::Personality)            && let Some(personality) = turn_context.personality        {            let model_info = turn_context.model_info.clone();            let has_baked_personality = model_info.supports_personality()                && base_instructions == model_info.get_model_instructions(Some(personality));            if !has_baked_personality                && let Some(personality_message) =                    crate::context_manager::updates::personality_message_for(                        &model_info,                        personality,                    )            {                developer_sections                    .push(PersonalitySpecInstructions::new(personality_message).render());            }        }        if turn_context.config.include_apps_instructions && turn_context.apps_enabled() {            let mcp_connection_manager = self.services.mcp_connection_manager.load_full();            let accessible_and_enabled_connectors =                connectors::list_accessible_and_enabled_connectors_from_manager(                    &mcp_connection_manager,                    &turn_context.config,                )                .await;            if let Some(apps_instructions) =                AppsInstructions::from_connectors(&accessible_and_enabled_connectors)            {                developer_sections.push(apps_instructions.render());            }        }        if turn_context.config.include_skill_instructions {            let available_skills = build_available_skills(                &turn_context.turn_skills.outcome,                default_skill_metadata_budget(turn_context.model_info.context_window),                SkillRenderSideEffects::ThreadStart {                    session_telemetry: &self.services.session_telemetry,                },            );            if let Some(available_skills) = available_skills {                let warning_message = available_skills.warning_message.clone();                let skills_instructions = AvailableSkillsInstructions::from(available_skills);                if let Some(warning_message) = warning_message {                    self.send_event_raw(Event {                        id: String::new(),                        msg: EventMsg::Warning(WarningEvent {                            message: warning_message,                        }),                    })                    .await;                }                developer_sections.push(skills_instructions.render());            }        }        let loaded_plugins = self            .services            .plugins_manager            .plugins_for_config(&turn_context.config.plugins_config_input())            .await;        if let Some(plugin_instructions) =            AvailablePluginsInstructions::from_plugins(loaded_plugins.capability_summaries())        {            developer_sections.push(plugin_instructions.render());        }        let context_contributors = self.services.extensions.context_contributors().to_vec();        for contributor in context_contributors {            for fragment in contributor                .contribute(                    &self.services.session_extension_data,                    &self.services.thread_extension_data,                )                .await            {                match fragment.slot() {                    PromptSlot::DeveloperPolicy | PromptSlot::DeveloperCapabilities => {                        developer_sections.push(fragment.text().to_string());                    }                    PromptSlot::ContextualUser => {                        contextual_user_sections.push(fragment.text().to_string());                    }                    PromptSlot::SeparateDeveloper => {                        separate_developer_sections.push(fragment.text().to_string());                    }                }            }        }        if let Some(user_instructions) = turn_context.user_instructions.as_deref() {            contextual_user_sections.push(user_instructions.to_string());        }        // This is full-context metadata. Steady-state context diffs should not re-emit it.        if turn_context.features.enabled(Feature::TokenBudget)            && let Some(model_context_window) = turn_context.model_context_window()        {            developer_sections.push(                crate::context::TokenBudgetContext::new(                    self.thread_id(),                    auto_compact_window_id,                    model_context_window,                )                .render(),            );        }        if turn_context.config.include_environment_context {            let shell = self.user_shell();            let subagents = self                .services                .agent_control                .format_environment_context_subagents(self.thread_id)                .await;            contextual_user_sections.push(                crate::context::EnvironmentContext::from_turn_context(turn_context, shell.as_ref())                    .with_subagents(subagents)                    .render(),            );        }        let multi_agent_v2_usage_hint_text =            multi_agents::usage_hint_text(turn_context, &session_source);        let mut items = Vec::with_capacity(4);        if let Some(developer_message) =            crate::context_manager::updates::build_developer_update_item(developer_sections)        {            items.push(developer_message);        }        for section in separate_developer_sections {            if let Some(developer_message) =                crate::context_manager::updates::build_developer_update_item(vec![section])            {                items.push(developer_message);            }        }        if let Some(usage_hint_text) = multi_agent_v2_usage_hint_text            && let Some(usage_hint_message) =                crate::context_manager::updates::build_developer_update_item(vec![                    usage_hint_text.to_string(),                ])        {            items.push(usage_hint_message);        }        if let Some(contextual_user_message) =            crate::context_manager::updates::build_contextual_user_message(contextual_user_sections)        {            items.push(contextual_user_message);        }        // Emit the guardian policy prompt as a separate developer item so the guardian        // subagent sees a distinct, easy-to-audit instruction block.        if separate_guardian_developer_message            && let Some(developer_instructions) = turn_context.developer_instructions.as_deref()            && !developer_instructions.is_empty()            && let Some(guardian_developer_message) =                crate::context_manager::updates::build_developer_update_item(vec![                    developer_instructions.to_string(),                ])        {            items.push(guardian_developer_message);        }        items    }    pub(crate) async fn persist_rollout_items(&self, items: &[RolloutItem]) {        if let Some(live_thread) = self.live_thread()            && let Err(e) = live_thread.append_items(items).await        {            error!("failed to record rollout items: {e:#}");        }    }    pub(crate) async fn clone_history(&self) -> ContextManager {        let state = self.state.lock().await;        state.clone_history()    }    pub(crate) async fn current_window_id(&self) -> String {        let state = self.state.lock().await;        let thread_id = self.thread_id;        let window_id = state.auto_compact_window_id();        format!("{thread_id}:{window_id}")    }    pub(crate) async fn advance_auto_compact_window_id(&self) -> u64 {        let mut state = self.state.lock().await;        state.advance_auto_compact_window_id()    }    pub(crate) async fn request_new_context_window(&self) {        let mut state = self.state.lock().await;        state.request_new_context_window();    }    pub(crate) async fn maybe_start_new_context_window(        &self,        turn_context: &TurnContext,    ) -> Option<u64> {        let window_id = {            let mut state = self.state.lock().await;            state.start_new_context_window_if_requested()        };        let window_id = window_id?;        let context_items = self.build_initial_context(turn_context).await;        let turn_context_item = turn_context.to_turn_context_item();        let replacement_history = context_items;        {            let mut state = self.state.lock().await;            state.replace_history(replacement_history.clone(), Some(turn_context_item.clone()));        };        self.persist_rollout_items(&[            RolloutItem::Compacted(CompactedItem {                message: String::new(),                replacement_history: Some(replacement_history),                window_id: Some(window_id),            }),            RolloutItem::TurnContext(turn_context_item),        ])        .await;        {            let mut state = self.state.lock().await;            state.queue_pending_session_start_source(codex_hooks::SessionStartSource::Compact);        }        self.recompute_token_usage(turn_context).await;        Some(window_id)    }    pub(crate) async fn reference_context_item(&self) -> Option<TurnContextItem> {        let state = self.state.lock().await;        state.reference_context_item()    }    /// Persist the latest turn context snapshot for the first real user turn and for    /// steady-state turns that emit model-visible context updates.    ///    /// When the reference snapshot is missing, this injects full initial context. Otherwise, it    /// emits only settings diff items.    ///    /// If full context is injected and a model switch occurred, this prepends the    /// `<model_switch>` developer message so model-specific instructions are not lost.    ///    /// This is the normal runtime path that establishes a new `reference_context_item`.    /// Mid-turn compaction is the other path that can re-establish that baseline when it    /// reinjects full initial context into replacement history. Other non-regular tasks    /// intentionally do not update the baseline.    #[instrument(level = "trace", skip_all)]    pub(crate) async fn record_context_updates_and_set_reference_context_item(        &self,        turn_context: &TurnContext,    ) {        let reference_context_item = {            let state = self.state.lock().await;            state.reference_context_item()        };        let should_inject_full_context = reference_context_item.is_none();        let context_items = if should_inject_full_context {            self.build_initial_context(turn_context).await        } else {            // Steady-state path: append only context diffs to minimize token overhead.            self.build_settings_update_items(reference_context_item.as_ref(), turn_context)                .await        };        let turn_context_item = turn_context.to_turn_context_item();        if !context_items.is_empty() {            self.record_conversation_items(turn_context, &context_items)                .await;        }        // Persist one `TurnContextItem` per real user turn so resume/lazy replay can recover the        // latest durable baseline even when this turn emitted no model-visible context diffs.        self.persist_rollout_items(&[RolloutItem::TurnContext(turn_context_item.clone())])            .await;        // Advance the in-memory diff baseline even when this turn emitted no model-visible        // context items. This keeps later runtime diffing aligned with the current turn state.        let mut state = self.state.lock().await;        state.set_reference_context_item(Some(turn_context_item));    }    pub(crate) async fn update_token_usage_info(        &self,        turn_context: &TurnContext,        token_usage: Option<&TokenUsage>,    ) {        self.record_token_usage_info(turn_context, token_usage)            .await;        self.send_token_count_event(turn_context).await;    }    pub(crate) async fn record_token_usage_info(        &self,        turn_context: &TurnContext,        token_usage: Option<&TokenUsage>,    ) {        if let Some(token_usage) = token_usage {            let token_info = {                let mut state = self.state.lock().await;                state                    .update_token_info_from_usage(token_usage, turn_context.model_context_window());                if matches!(                    turn_context.config.model_auto_compact_token_limit_scope,                    AutoCompactTokenLimitScope::BodyAfterPrefix                ) {                    state.ensure_auto_compact_window_server_prefill_from_usage(token_usage);                }                state.token_info()            };            if let Some(token_info) = token_info.as_ref() {                for contributor in self.services.extensions.token_usage_contributors() {                    contributor                        .on_token_usage(                            &self.services.session_extension_data,                            &self.services.thread_extension_data,                            turn_context.extension_data.as_ref(),                            token_info,                        )                        .await;                }            }        }    }    pub(crate) async fn recompute_token_usage(&self, turn_context: &TurnContext) {        let history = self.clone_history().await;        let base_instructions = self.get_base_instructions().await;        let Some(estimated_total_tokens) =            history.estimate_token_count_with_base_instructions(&base_instructions)        else {            return;        };        {            let mut state = self.state.lock().await;            let mut info = state.token_info().unwrap_or(TokenUsageInfo {                total_token_usage: TokenUsage::default(),                last_token_usage: TokenUsage::default(),                model_context_window: None,            });            info.last_token_usage = TokenUsage {                input_tokens: 0,                cached_input_tokens: 0,                output_tokens: 0,                reasoning_output_tokens: 0,                total_tokens: estimated_total_tokens.max(0),            };            if let Some(model_context_window) = turn_context.model_context_window() {                info.model_context_window = Some(model_context_window);            }            state.set_token_info(Some(info));        }        self.set_auto_compact_window_estimated_prefill_for_scope(            turn_context,            estimated_total_tokens,        )        .await;        self.send_token_count_event(turn_context).await;    }    pub(crate) async fn update_rate_limits(        &self,        turn_context: &TurnContext,        new_rate_limits: RateLimitSnapshot,    ) {        self.record_rate_limits_info(new_rate_limits).await;        self.send_token_count_event(turn_context).await;    }    pub(crate) async fn record_rate_limits_info(&self, new_rate_limits: RateLimitSnapshot) {        {            let mut state = self.state.lock().await;            state.set_rate_limits(new_rate_limits);        }    }    pub(crate) async fn mcp_dependency_prompted(&self) -> HashSet<String> {        let state = self.state.lock().await;        state.mcp_dependency_prompted()    }    pub(crate) async fn record_mcp_dependency_prompted<I>(&self, names: I)    where        I: IntoIterator<Item = String>,    {        let mut state = self.state.lock().await;        state.record_mcp_dependency_prompted(names);    }    pub(crate) async fn set_server_reasoning_included(&self, included: bool) {        let mut state = self.state.lock().await;        state.set_server_reasoning_included(included);    }    pub(crate) async fn send_token_count_event(&self, turn_context: &TurnContext) {        let (info, rate_limits) = {            let state = self.state.lock().await;            state.token_info_and_rate_limits()        };        let event = EventMsg::TokenCount(TokenCountEvent { info, rate_limits });        self.send_event(turn_context, event).await;    }    pub(crate) async fn set_total_tokens_full(&self, turn_context: &TurnContext) {        if let Some(context_window) = turn_context.model_context_window() {            let mut state = self.state.lock().await;            state.set_token_usage_full(context_window);        }        self.send_token_count_event(turn_context).await;    }    pub(crate) async fn record_response_item_and_emit_turn_item(        &self,        turn_context: &TurnContext,        response_item: ResponseItem,    ) {        // Add to conversation history and persist response item to rollout.        self.record_conversation_items(turn_context, std::slice::from_ref(&response_item))            .await;        // Derive a turn item and emit lifecycle events if applicable.        if let Some(item) = parse_turn_item(&response_item) {            self.emit_turn_item_started(turn_context, &item).await;            self.emit_turn_item_completed(turn_context, item).await;        }    }    pub(crate) async fn record_user_prompt_and_emit_turn_item(        &self,        turn_context: &TurnContext,        input: &[UserInput],        client_id: Option<String>,    ) {        // Persist the user message to history, but emit the turn item from `UserInput` so        // UI-only `text_elements` are preserved. `ResponseItem::Message` does not carry        // those spans, and `record_response_item_and_emit_turn_item` would drop them.        let response_item = self.response_item_from_user_input(turn_context, input.to_vec());        self.record_conversation_items(turn_context, std::slice::from_ref(&response_item))            .await;        let mut user_message_item = UserMessageItem::new(input);        user_message_item.client_id = client_id;        let turn_item = TurnItem::UserMessage(user_message_item);        self.emit_turn_item_started(turn_context, &turn_item).await;        self.emit_turn_item_completed(turn_context, turn_item).await;        self.ensure_rollout_materialized().await;    }    pub(crate) async fn notify_stream_error(        &self,        turn_context: &TurnContext,        message: impl Into<String>,        codex_error: CodexErr,    ) {        let additional_details = codex_error.to_string();        let codex_error_info = CodexErrorInfo::ResponseStreamDisconnected {            http_status_code: codex_error.http_status_code_value(),        };        let event = EventMsg::StreamError(StreamErrorEvent {            message: message.into(),            codex_error_info: Some(codex_error_info),            additional_details: Some(additional_details),        });        self.send_event(turn_context, event).await;    }    /// Inject additional user input into the currently active turn.    ///    /// Returns the active turn id when accepted.    #[expect(        clippy::await_holding_invalid_type,        reason = "active turn checks and turn state updates must remain atomic"    )]    pub async fn steer_input(        &self,        input: Vec<UserInput>,        additional_context: BTreeMap<String, AdditionalContextEntry>,        expected_turn_id: Option<&str>,        client_user_message_id: Option<String>,        responsesapi_client_metadata: Option<HashMap<String, String>>,    ) -> Result<String, SteerInputError> {        let mut active = self.active_turn.lock().await;        let Some(active_turn) = active.as_mut() else {            return Err(SteerInputError::NoActiveTurn(input));        };        let Some(active_task) = active_turn.task.as_ref() else {            return Err(SteerInputError::NoActiveTurn(input));        };        let active_turn_id = &active_task.turn_context.sub_id;        if let Some(expected_turn_id) = expected_turn_id            && expected_turn_id != active_turn_id        {            return Err(SteerInputError::ExpectedTurnMismatch {                expected: expected_turn_id.to_string(),                actual: active_turn_id.clone(),            });        }        match active_task.kind {            crate::state::TaskKind::Regular => {}            crate::state::TaskKind::Review => {                return Err(SteerInputError::ActiveTurnNotSteerable {                    turn_kind: NonSteerableTurnKind::Review,                });            }            crate::state::TaskKind::Compact => {                return Err(SteerInputError::ActiveTurnNotSteerable {                    turn_kind: NonSteerableTurnKind::Compact,                });            }        }        if input.is_empty() {            return Err(SteerInputError::EmptyInput);        }        let additional_context_input = {            let mut state = self.state.lock().await;            state.additional_context.merge(additional_context)        };        if let Some(responsesapi_client_metadata) = responsesapi_client_metadata {            active_task                .turn_context                .turn_metadata_state                .set_responsesapi_client_metadata(responsesapi_client_metadata);        }        let mut pending_input = additional_context_input            .into_iter()            .map(ResponseItem::from)            .map(TurnInput::ResponseItem)            .collect::<Vec<_>>();        pending_input.push(TurnInput::UserInput {            content: input,            client_id: client_user_message_id,        });        self.input_queue            .extend_pending_input_and_accept_mailbox_delivery_for_turn_state(                active_turn.turn_state.as_ref(),                pending_input,            )            .await;        Ok(active_turn_id.clone())    }    pub(crate) async fn record_memory_citation_for_turn(&self, sub_id: &str) {        let turn_state = self            .input_queue            .turn_state_for_sub_id(&self.active_turn, sub_id)            .await;        let Some(turn_state) = turn_state else {            return;        };        turn_state.lock().await.has_memory_citation = true;    }    pub async fn interrupt_task(self: &Arc<Self>) {        info!("interrupt received: abort current task, if any");        let had_active_turn = self.active_turn.lock().await.is_some();        self.abort_all_tasks(TurnAbortReason::Interrupted).await;        if !had_active_turn {            self.cancel_mcp_startup().await;        }    }    pub(crate) fn hooks(&self) -> Arc<Hooks> {        self.services.hooks.load_full()    }    pub(crate) fn user_shell(&self) -> Arc<shell::Shell> {        Arc::clone(&self.services.user_shell)    }    pub(crate) async fn current_rollout_path(&self) -> anyhow::Result<Option<PathBuf>> {        let Some(live_thread) = self.live_thread() else {            return Ok(None);        };        live_thread.local_rollout_path().await.map_err(Into::into)    }    pub(crate) async fn hook_transcript_path(&self) -> Option<PathBuf> {        self.ensure_rollout_materialized().await;        match self.current_rollout_path().await {            Ok(path) => path,            Err(err) => {                warn!("{err}");                None            }        }    }    pub(crate) async fn take_pending_session_start_source(        &self,    ) -> Option<codex_hooks::SessionStartSource> {        let mut state = self.state.lock().await;        state.take_pending_session_start_source()    }    fn show_raw_agent_reasoning(&self) -> bool {        self.services.show_raw_agent_reasoning    }}pub(crate) fn emit_subagent_session_started(    analytics_events_client: &AnalyticsEventsClient,    client_metadata: AppServerClientMetadata,    session_id: SessionId,    thread_id: ThreadId,    parent_thread_id: Option<ThreadId>,    thread_config: ThreadConfigSnapshot,    subagent_source: SubAgentSource,) {    let AppServerClientMetadata {        client_name,        client_version,    } = client_metadata;    let (Some(client_name), Some(client_version)) = (client_name, client_version) else {        tracing::warn!("skipping subagent thread analytics: missing inherited client metadata");        return;    };    let created_at = SystemTime::now()        .duration_since(UNIX_EPOCH)        .unwrap_or_default()        .as_secs();    analytics_events_client.track_subagent_thread_started(SubAgentThreadStartedInput {        session_id: session_id.to_string(),        thread_id: thread_id.to_string(),        parent_thread_id: parent_thread_id.map(|thread_id| thread_id.to_string()),        forked_from_thread_id: thread_config            .forked_from_thread_id            .map(|thread_id| thread_id.to_string()),        product_client_id: client_name.clone(),        client_name,        client_version,        model: thread_config.model,        ephemeral: thread_config.ephemeral,        subagent_source,        created_at,    });}/// Builds the hook engine for one config snapshot, including any enabled plugin hooks.async fn build_hooks_for_config(    config: &Config,    plugins_manager: &PluginsManager,    environment: Option<&TurnEnvironment>,) -> Hooks {    let (hook_shell_program, hook_shell_argv) = environment        .and_then(|environment| environment.shell.as_ref())        .map(|shell| {            let mut argv = shell.derive_exec_args("", /*use_login_shell*/ false);            let program = argv.remove(0);            let _ = argv.pop();            (Some(program), argv)        })        .unwrap_or_default();    let plugins_input = config.plugins_config_input();    let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await;    let plugin_hook_sources = plugin_outcome.effective_plugin_hook_sources();    let plugin_hook_load_warnings = plugin_outcome.effective_plugin_hook_warnings();    Hooks::new(HooksConfig {        legacy_notify_argv: config.notify.clone(),        feature_enabled: config.features.enabled(Feature::CodexHooks),        bypass_hook_trust: config.bypass_hook_trust,        config_layer_stack: Some(config.config_layer_stack.clone()),        plugin_hook_sources,        plugin_hook_load_warnings,        shell_program: hook_shell_program,        shell_args: hook_shell_argv,    })}#[cfg(test)]pub(crate) mod tests;